rest/src/lib.rs
// Apache-2.0 licensed. See LICENSE-APACHE.
//! An Iceberg-REST (OpenAPI) front over skade-katalog's [`RedbCatalog`] (axum).
//!
//! This is the **REST mode** of the tri-modal catalog (embedded · REST · gRPC):
//! the SAME `RedbCatalog` — same redb file, same write-lock, same
//! `commit_table` optimistic group-commit — served on the ecosystem-interop
//! wire so Spark / PyIceberg / DuckDB / Trino (any stock Iceberg-REST client)
//! can drive it. It was promoted verbatim out of the benchmark crate
//! (`bench/src/rest_shim.rs`) into this shipped library; `bench/` now depends on
//! it rather than carrying its own copy (REUSE, not twin).
//!
//! Scope: the endpoints an Iceberg-REST client exercises — `config`, namespace
//! create/list/get/exists, table create/load/list/head/drop/**commit**. Table
//! metadata is forwarded as the raw on-disk JSON (which is the canonical Iceberg
//! metadata document) — iceberg's `TableMetadata` is not `Serialize`, but the
//! file already is the wire form, so we splice its bytes into `LoadTableResult`
//! rather than re-serializing.
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use anyhow::Result;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::{Json, Router};
use iceberg::io::FileIO;
use iceberg::spec::Schema;
use iceberg::{
Catalog, ErrorKind, NamespaceIdent, TableCreation, TableIdent, TableRequirement, TableUpdate,
};
use serde::Deserialize;
use serde_json::{json, Value};
use skade_katalog::RedbCatalog;
type Cat = Arc<RedbCatalog>;
/// Test-matrix emit shim. Compiled-out `#[inline]` no-op unless `--features
/// testmatrix` (mirrors `skade_katalog`'s own shim), so a normal build carries
/// neither the emit nor the `nornir-testmatrix` dependency.
#[inline]
fn functional_status(component: &str, check: &str, ok: bool, detail: &str) {
#[cfg(feature = "testmatrix")]
nornir_testmatrix::functional_status(component, check, ok, detail);
#[cfg(not(feature = "testmatrix"))]
{
let _ = (component, check, ok, detail);
}
}
/// Build the axum router that fronts `cat`. Shared by [`serve`] and [`spawn`].
pub fn router(cat: RedbCatalog) -> Router {
let state: Cat = Arc::new(cat);
Router::new()
.route("/v1/config", get(config))
.route("/v1/namespaces", get(list_namespaces).post(create_namespace))
// axum 0.8 captures are `{name}`; a `:name` segment panics at router build.
.route("/v1/namespaces/{ns}", get(get_namespace).head(head_namespace))
.route("/v1/namespaces/{ns}/tables", get(list_tables).post(create_table))
.route(
"/v1/namespaces/{ns}/tables/{table}",
get(load_table)
.head(head_table)
.delete(drop_table)
.post(commit_table),
)
.with_state(state)
}
/// Serve the REST front over `cat` on `addr` **until the process exits** (blocks).
/// This is the mode the `skade-katalog serve` binary uses for `--rest`.
pub async fn serve(cat: RedbCatalog, addr: SocketAddr) -> Result<()> {
let listener = tokio::net::TcpListener::bind(addr).await?;
axum::serve(listener, router(cat)).await?;
Ok(())
}
/// Bind + spawn the REST front in the background, returning the bound address
/// once it is listening. Used by tests and the benchmark harness.
pub async fn spawn(cat: RedbCatalog, addr: SocketAddr) -> Result<SocketAddr> {
let listener = tokio::net::TcpListener::bind(addr).await?;
let bound = listener.local_addr()?;
let app = router(cat);
tokio::spawn(async move {
let _ = axum::serve(listener, app).await;
});
Ok(bound)
}
// Namespace path segments are joined with the unit separator () per the
// Iceberg REST spec; single-level namespaces are just the bare name.
fn parse_ns(s: &str) -> NamespaceIdent {
let parts: Vec<String> = s.split('\u{1f}').map(|p| p.to_string()).collect();
NamespaceIdent::from_vec(parts).unwrap_or_else(|_| NamespaceIdent::new(s.to_string()))
}
/// Fetch the metadata JSON at `loc` **through the table's own `FileIO`**.
///
/// `FileIO` is the catalog's configured storage — `LocalFsStorageFactory` for a
/// `file://` warehouse (what this front has always served), or any other
/// `StorageFactory` the operator wired in (e.g. skade's `ObjectStoreFactory`
/// over S3/MinIO). Going through it is what lets the REST front serve a
/// warehouse whose blobs are NOT on this box's filesystem: an `s3://…`
/// metadata-location is read by the same storage the catalog writes with.
///
/// The historical `tokio::fs` read is kept as a fallback so nothing that worked
/// before can stop working: if the `FileIO` path errors (or the catalog was
/// built with a storage that cannot address this location), we still try the
/// plain local file the old code read.
async fn read_metadata_json(file_io: &FileIO, loc: &str) -> Result<Value> {
let via_io = async {
let bytes = file_io.new_input(loc)?.read().await?;
Ok::<Value, anyhow::Error>(serde_json::from_slice(&bytes)?)
}
.await;
match via_io {
Ok(v) => Ok(v),
Err(io_err) => {
let path = loc.strip_prefix("file://").unwrap_or(loc);
match tokio::fs::read(path).await {
Ok(bytes) => Ok(serde_json::from_slice(&bytes)?),
// Report the FileIO error — it is the one that describes the
// storage the catalog is actually configured with.
Err(_) => Err(io_err),
}
}
}
}
fn err(code: StatusCode, msg: &str) -> Response {
(code, Json(json!({"error": {"message": msg, "type": "Error", "code": code.as_u16()}}))).into_response()
}
async fn config() -> Response {
Json(json!({ "overrides": {}, "defaults": {} })).into_response()
}
#[derive(Deserialize)]
struct CreateNamespaceRequest {
namespace: Vec<String>,
#[serde(default)]
properties: HashMap<String, String>,
}
async fn create_namespace(State(cat): State<Cat>, Json(req): Json<CreateNamespaceRequest>) -> Response {
let ns = match NamespaceIdent::from_vec(req.namespace.clone()) {
Ok(n) => n,
Err(_) => return err(StatusCode::BAD_REQUEST, "invalid namespace"),
};
match cat.create_namespace(&ns, req.properties.clone()).await {
Ok(_) => Json(json!({ "namespace": req.namespace, "properties": req.properties })).into_response(),
Err(e) => err(StatusCode::CONFLICT, &e.to_string()),
}
}
async fn head_namespace(State(cat): State<Cat>, Path(ns): Path<String>) -> StatusCode {
match cat.namespace_exists(&parse_ns(&ns)).await {
Ok(true) => StatusCode::NO_CONTENT,
_ => StatusCode::NOT_FOUND,
}
}
async fn get_namespace(State(cat): State<Cat>, Path(ns): Path<String>) -> Response {
let nsi = parse_ns(&ns);
match cat.get_namespace(&nsi).await {
Ok(n) => Json(json!({ "namespace": nsi.inner(), "properties": n.properties() })).into_response(),
Err(e) => err(StatusCode::NOT_FOUND, &e.to_string()),
}
}
async fn list_namespaces(State(cat): State<Cat>) -> Response {
match cat.list_namespaces(None).await {
Ok(list) => {
let ns: Vec<Vec<String>> = list.into_iter().map(|n| n.inner()).collect();
Json(json!({ "namespaces": ns })).into_response()
}
Err(e) => err(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
}
}
async fn list_tables(State(cat): State<Cat>, Path(ns): Path<String>) -> Response {
let nsi = parse_ns(&ns);
match cat.list_tables(&nsi).await {
Ok(idents) => {
let ids: Vec<Value> = idents
.into_iter()
.map(|t| json!({ "namespace": t.namespace().clone().inner(), "name": t.name() }))
.collect();
Json(json!({ "identifiers": ids })).into_response()
}
Err(e) => err(StatusCode::NOT_FOUND, &e.to_string()),
}
}
#[derive(Deserialize)]
struct CreateTableRequest {
name: String,
schema: Schema,
#[serde(default)]
location: Option<String>,
// partition-spec / write-order / stage-create / properties are accepted and
// ignored for the benchmark workload.
}
async fn create_table(State(cat): State<Cat>, Path(ns): Path<String>, Json(req): Json<CreateTableRequest>) -> Response {
let nsi = parse_ns(&ns);
// The TableCreation builder is typestate (bon) — branch rather than reassign.
let creation = match req.location {
Some(loc) => TableCreation::builder().name(req.name).schema(req.schema).location(loc).build(),
None => TableCreation::builder().name(req.name).schema(req.schema).build(),
};
match cat.create_table(&nsi, creation).await {
Ok(table) => load_table_result(&table).await,
Err(e) => err(StatusCode::CONFLICT, &e.to_string()),
}
}
async fn load_table(State(cat): State<Cat>, Path((ns, table)): Path<(String, String)>) -> Response {
let ident = TableIdent::new(parse_ns(&ns), table);
match cat.load_table(&ident).await {
Ok(t) => load_table_result(&t).await,
Err(e) => err(StatusCode::NOT_FOUND, &e.to_string()),
}
}
async fn load_table_result(table: &iceberg::table::Table) -> Response {
let loc = table.metadata_location().unwrap_or_default().to_string();
match read_metadata_json(table.file_io(), &loc).await {
Ok(meta) => Json(json!({ "metadata-location": loc, "metadata": meta, "config": {} })).into_response(),
Err(e) => err(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
}
}
async fn head_table(State(cat): State<Cat>, Path((ns, table)): Path<(String, String)>) -> StatusCode {
let ident = TableIdent::new(parse_ns(&ns), table);
match cat.table_exists(&ident).await {
Ok(true) => StatusCode::NO_CONTENT,
_ => StatusCode::NOT_FOUND,
}
}
async fn drop_table(State(cat): State<Cat>, Path((ns, table)): Path<(String, String)>) -> StatusCode {
let ident = TableIdent::new(parse_ns(&ns), table);
match cat.drop_table(&ident).await {
Ok(_) => StatusCode::NO_CONTENT,
Err(_) => StatusCode::NOT_FOUND,
}
}
// Iceberg-REST commit: apply the request's `requirements` + `updates` against
// the backing `RedbCatalog` and return a `CommitTableResponse`. We can't
// deserialize the wire body into iceberg's `CommitTableRequest`/`TableCommit`
// (the `TableCommit` builder is crate-private to iceberg), so the body is
// captured as the raw `requirements`/`updates` vectors and handed to
// `RedbCatalog::commit_table`, which mirrors `TableCommit::apply` + persists
// through the optimistic group-commit path. The response splices the freshly
// written on-disk metadata JSON in place of a (non-`Serialize`) `TableMetadata`,
// exactly as `load_table_result` does.
#[derive(Deserialize)]
struct CommitTableRequest {
#[serde(default)]
requirements: Vec<TableRequirement>,
#[serde(default)]
updates: Vec<TableUpdate>,
}
async fn commit_table(
State(cat): State<Cat>,
Path((ns, table)): Path<(String, String)>,
Json(req): Json<CommitTableRequest>,
) -> Response {
let ident = TableIdent::new(parse_ns(&ns), table);
match cat.commit_table(ident, req.requirements, req.updates).await {
Ok(t) => {
functional_status("skade-katalog-rest", "commit_table", true, "committed via REST");
commit_table_result(&t).await
}
Err(e) => {
let code = match e.kind() {
ErrorKind::TableNotFound => StatusCode::NOT_FOUND,
ErrorKind::CatalogCommitConflicts => StatusCode::CONFLICT,
_ => StatusCode::INTERNAL_SERVER_ERROR,
};
functional_status(
"skade-katalog-rest",
"commit_table",
false,
&format!("rejected {}", code.as_u16()),
);
err(code, &e.to_string())
}
}
}
// `CommitTableResponse` is `{ metadata-location, metadata }` — same metadata
// JSON-splice trick as `load_table_result`, minus the `config` field.
async fn commit_table_result(table: &iceberg::table::Table) -> Response {
let loc = table.metadata_location().unwrap_or_default().to_string();
match read_metadata_json(table.file_io(), &loc).await {
Ok(meta) => Json(json!({ "metadata-location": loc, "metadata": meta })).into_response(),
Err(e) => err(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),
}
}