bench/src/grpc_catalog.rs
//! An `iceberg::Catalog` adapter over the skade-katalog **gRPC** front, so the
//! benchmark `scenarios::*` run the *identical code* against the gRPC wire that
//! they run embedded and REST-fronted (the third of the three fronts).
//!
//! It delegates every catalog op to the generated `skade.katalog.v1` tonic
//! client, and reconstructs `iceberg::table::Table` handles the same way
//! iceberg's own `MemoryCatalog` does — reading the canonical metadata document
//! from the location the front returns (`TableMetadata::read_from`) over a local
//! FileIO. Only the methods the bench scenarios exercise are implemented; the
//! mutation methods the trait also requires (`update_table`, `rename_table`, …)
//! return `FeatureUnsupported` — the gRPC arm times reads/creates, not those.
use std::collections::HashMap;
use anyhow::Result;
use async_trait::async_trait;
use iceberg::io::FileIO;
use iceberg::spec::TableMetadata;
use iceberg::table::Table;
use iceberg::{
Catalog, Error, ErrorKind, Namespace, NamespaceIdent, Result as IceResult, TableCommit,
TableCreation, TableIdent,
};
use skade_katalog_grpc::pb;
use tonic::transport::Channel;
type Client = pb::catalog_client::CatalogClient<Channel>;
/// A catalog client that speaks skade-katalog gRPC, presented as an
/// `iceberg::Catalog` so the generic bench scenarios drive it unchanged.
#[derive(Debug)]
pub struct GrpcCatalog {
client: Client,
file_io: FileIO,
}
impl GrpcCatalog {
/// Connect to a running `skade.katalog.v1.Catalog` gRPC server at `addr`.
pub async fn connect(addr: std::net::SocketAddr) -> Result<Self> {
// TCP_NODELAY on the client too — the request side is tiny so it matters
// less than the server, but keep both ends consistent (no Nagle stalls).
let channel = tonic::transport::Endpoint::from_shared(format!("http://{addr}"))?
.tcp_nodelay(true)
.connect()
.await?;
let client = Client::new(channel);
Ok(Self { client, file_io: FileIO::new_with_fs() })
}
async fn table_from(&self, ident: TableIdent, loc: String) -> IceResult<Table> {
let metadata = TableMetadata::read_from(&self.file_io, &loc).await?;
Table::builder()
.identifier(ident)
.metadata(metadata)
.metadata_location(loc)
.file_io(self.file_io.clone())
.runtime(iceberg::Runtime::try_current()?)
.build()
}
}
/// Map a tonic status back to an iceberg error kind (inverse of the front's
/// `to_status`): NOT_FOUND → TableNotFound, ABORTED → CatalogCommitConflicts.
fn st(e: tonic::Status) -> Error {
let kind = match e.code() {
tonic::Code::NotFound => ErrorKind::TableNotFound,
tonic::Code::Aborted => ErrorKind::CatalogCommitConflicts,
_ => ErrorKind::Unexpected,
};
Error::new(kind, e.message().to_string())
}
fn unsupported(op: &str) -> Error {
Error::new(ErrorKind::FeatureUnsupported, format!("GrpcCatalog bench arm: {op} not wired"))
}
#[async_trait]
impl Catalog for GrpcCatalog {
async fn list_namespaces(&self, parent: Option<&NamespaceIdent>) -> IceResult<Vec<NamespaceIdent>> {
let mut c = self.client.clone();
let parent = parent.map(|p| p.clone().inner()).unwrap_or_default();
let list = c
.list_namespaces(pb::ListNamespacesRequest { parent })
.await
.map_err(st)?
.into_inner();
list.namespaces
.into_iter()
.map(|n| NamespaceIdent::from_vec(n.levels))
.collect()
}
async fn create_namespace(
&self,
namespace: &NamespaceIdent,
properties: HashMap<String, String>,
) -> IceResult<Namespace> {
let mut c = self.client.clone();
let r = c
.create_namespace(pb::CreateNamespaceReq {
levels: namespace.clone().inner(),
properties,
})
.await
.map_err(st)?
.into_inner();
Ok(Namespace::with_properties(namespace.clone(), r.properties))
}
async fn get_namespace(&self, namespace: &NamespaceIdent) -> IceResult<Namespace> {
let mut c = self.client.clone();
let r = c
.get_namespace(pb::NamespaceIdentMsg { levels: namespace.clone().inner() })
.await
.map_err(st)?
.into_inner();
Ok(Namespace::with_properties(namespace.clone(), r.properties))
}
async fn namespace_exists(&self, namespace: &NamespaceIdent) -> IceResult<bool> {
let mut c = self.client.clone();
let r = c
.namespace_exists(pb::NamespaceIdentMsg { levels: namespace.clone().inner() })
.await
.map_err(st)?;
Ok(r.into_inner().exists)
}
async fn update_namespace(&self, _ns: &NamespaceIdent, _props: HashMap<String, String>) -> IceResult<()> {
Err(unsupported("update_namespace"))
}
async fn drop_namespace(&self, _namespace: &NamespaceIdent) -> IceResult<()> {
Err(unsupported("drop_namespace"))
}
async fn list_tables(&self, namespace: &NamespaceIdent) -> IceResult<Vec<TableIdent>> {
let mut c = self.client.clone();
let r = c
.list_tables(pb::NamespaceIdentMsg { levels: namespace.clone().inner() })
.await
.map_err(st)?
.into_inner();
r.tables
.into_iter()
.map(|t| Ok(TableIdent::new(NamespaceIdent::from_vec(t.namespace)?, t.name)))
.collect()
}
async fn create_table(&self, namespace: &NamespaceIdent, creation: TableCreation) -> IceResult<Table> {
let schema_json = serde_json::to_string(&creation.schema)
.map_err(|e| Error::new(ErrorKind::DataInvalid, e.to_string()))?;
let mut c = self.client.clone();
let r = c
.create_table(pb::CreateTableReq {
namespace: namespace.clone().inner(),
name: creation.name.clone(),
schema_json,
location: creation.location.clone().unwrap_or_default(),
})
.await
.map_err(st)?
.into_inner();
let ident = TableIdent::new(namespace.clone(), creation.name);
self.table_from(ident, r.metadata_location).await
}
async fn load_table(&self, table: &TableIdent) -> IceResult<Table> {
let mut c = self.client.clone();
let r = c
.load_table(pb::TableIdentMsg {
namespace: table.namespace().clone().inner(),
name: table.name().to_string(),
})
.await
.map_err(st)?
.into_inner();
self.table_from(table.clone(), r.metadata_location).await
}
async fn drop_table(&self, table: &TableIdent) -> IceResult<()> {
let mut c = self.client.clone();
c.drop_table(pb::TableIdentMsg {
namespace: table.namespace().clone().inner(),
name: table.name().to_string(),
})
.await
.map_err(st)?;
Ok(())
}
/// The gRPC catalog protocol has no purge verb; required by iceberg-rust
/// main's `Catalog` since the 2026-10-02 rebase. Refused by name.
async fn purge_table(&self, table: &TableIdent) -> IceResult<()> {
Err(Error::new(
ErrorKind::FeatureUnsupported,
format!("grpc catalog has no purge_table ({table}); use drop_table"),
))
}
async fn table_exists(&self, table: &TableIdent) -> IceResult<bool> {
let mut c = self.client.clone();
let r = c
.table_exists(pb::TableIdentMsg {
namespace: table.namespace().clone().inner(),
name: table.name().to_string(),
})
.await
.map_err(st)?;
Ok(r.into_inner().exists)
}
async fn rename_table(&self, _src: &TableIdent, _dest: &TableIdent) -> IceResult<()> {
Err(unsupported("rename_table"))
}
async fn register_table(&self, _table: &TableIdent, _metadata_location: String) -> IceResult<Table> {
Err(unsupported("register_table"))
}
async fn update_table(&self, _commit: TableCommit) -> IceResult<Table> {
// The gRPC front carries requirements+updates as JSON; the trait hands us
// a `TableCommit` whose fields are crate-private to iceberg, so we can't
// repackage it here. The bench scenarios don't call `update_table`.
Err(unsupported("update_table"))
}
}