grpc/tests/client_roundtrip.rs
//! The **client** half against the server, in-process: `client::GrpcCatalog`
//! as an `iceberg::Catalog` over the native `skade.katalog.v1.Catalog` front,
//! held to what the embedded `RedbCatalog` behind the front says.
use std::collections::HashMap;
use std::sync::Arc;
use iceberg::io::LocalFsStorageFactory;
use iceberg::spec::{
FormatVersion, NestedField, PrimitiveType, Schema, Transform, Type, UnboundPartitionField, UnboundPartitionSpec,
};
use iceberg::transaction::{ApplyTransactionAction, Transaction};
use iceberg::{Catalog, CatalogBuilder, ErrorKind, NamespaceIdent, TableCreation, TableIdent};
use skade_katalog::{RedbCatalog, RedbCatalogBuilder};
use skade_katalog_grpc::client::GrpcCatalog;
use tempfile::TempDir;
fn schema() -> Schema {
Schema::builder()
.with_schema_id(0)
.with_fields(vec![NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into()])
.build()
.unwrap()
}
async fn embedded(tmp: &TempDir) -> RedbCatalog {
let warehouse = tmp.path().join("warehouse");
std::fs::create_dir_all(&warehouse).unwrap();
RedbCatalogBuilder::default()
.db_path(tmp.path().join("catalog.redb"))
.warehouse_location(format!("file://{}", warehouse.display()))
.with_storage_factory(Arc::new(LocalFsStorageFactory))
.load("skade", HashMap::new())
.await
.unwrap()
}
/// The front over `cat` and a client connected to it.
async fn front(cat: RedbCatalog) -> GrpcCatalog {
let bound = skade_katalog_grpc::spawn(cat, "127.0.0.1:0".parse().unwrap()).await.unwrap();
GrpcCatalog::connect(&format!("http://{bound}"), Arc::new(LocalFsStorageFactory)).await.unwrap()
}
#[tokio::test]
async fn the_client_is_an_iceberg_catalog_over_the_front() {
let tmp = TempDir::new().unwrap();
let cat = embedded(&tmp).await;
let grpc = front(cat.clone()).await;
// Namespaces.
let ns = NamespaceIdent::new("db".into());
let props = HashMap::from([("owner".to_string(), "bifrost".to_string())]);
grpc.create_namespace(&ns, props.clone()).await.unwrap();
assert_eq!(grpc.create_namespace(&ns, HashMap::new()).await.unwrap_err().kind(), ErrorKind::NamespaceAlreadyExists);
assert!(grpc.namespace_exists(&ns).await.unwrap());
assert_eq!(grpc.get_namespace(&ns).await.unwrap().properties().get("owner"), Some(&"bifrost".to_string()));
assert_eq!(grpc.list_namespaces(None).await.unwrap(), vec![ns.clone()]);
// A table: created over the wire, loaded back equal to the embedded load.
let creation = TableCreation::builder().name("t".into()).schema(schema()).build();
let created = grpc.create_table(&ns, creation).await.unwrap();
let ident = TableIdent::new(ns.clone(), "t".into());
assert_eq!(grpc.list_tables(&ns).await.unwrap(), vec![ident.clone()]);
let local = cat.load_table(&ident).await.unwrap();
let loaded = grpc.load_table(&ident).await.unwrap();
assert_eq!(loaded.metadata_location(), local.metadata_location());
assert_eq!(loaded.metadata(), local.metadata());
assert_eq!(created.metadata(), local.metadata());
assert_eq!(loaded.identifier(), &ident);
// A commit through `update_table` (CommitTable): persisted, a new pointer.
let tx = Transaction::new(&loaded);
let tx = tx.update_table_properties().set("answer".into(), "42".into()).apply(tx).unwrap();
let committed = tx.commit(&grpc).await.unwrap();
assert_ne!(committed.metadata_location(), loaded.metadata_location());
let local = cat.load_table(&ident).await.unwrap();
assert_eq!(local.metadata().properties().get("answer").map(String::as_str), Some("42"));
assert_eq!(committed.metadata(), local.metadata());
// Missing things keep their iceberg kinds.
let nobody = TableIdent::new(ns.clone(), "nobody".into());
assert_eq!(grpc.load_table(&nobody).await.unwrap_err().kind(), ErrorKind::TableNotFound);
assert!(!grpc.table_exists(&nobody).await.unwrap());
assert_eq!(
grpc.get_namespace(&NamespaceIdent::new("nowhere".into())).await.unwrap_err().kind(),
ErrorKind::NamespaceNotFound
);
// Drop: the pointer goes.
grpc.drop_table(&ident).await.unwrap();
assert!(!grpc.table_exists(&ident).await.unwrap());
}
#[tokio::test]
async fn what_the_wire_cannot_carry_is_refused_not_dropped() {
let tmp = TempDir::new().unwrap();
let grpc = front(embedded(&tmp).await).await;
let ns = NamespaceIdent::new("db".into());
grpc.create_namespace(&ns, HashMap::new()).await.unwrap();
let partitioned = TableCreation::builder()
.name("p".into())
.schema(schema())
.partition_spec(
UnboundPartitionSpec::builder()
.add_partition_field(
UnboundPartitionField::builder().source_ids(vec![1]).name("id").transform(Transform::Identity).build().unwrap(),
)
.unwrap()
.build(),
)
.build();
let with_props = TableCreation::builder().name("q".into()).schema(schema()).properties([("k".to_string(), "v".to_string())]).build();
let v3 = TableCreation::builder().name("r".into()).schema(schema()).format_version(FormatVersion::V3).build();
for (creation, what) in [(partitioned, "a partition spec"), (with_props, "table properties"), (v3, "a format version")] {
let name = creation.name.clone();
let e = grpc.create_table(&ns, creation).await.unwrap_err();
assert_eq!(e.kind(), ErrorKind::FeatureUnsupported, "{e}");
assert!(e.to_string().contains(what), "{e}");
assert!(!grpc.table_exists(&TableIdent::new(ns.clone(), name)).await.unwrap(), "nothing was created");
}
for e in [
grpc.rename_table(&TableIdent::new(ns.clone(), "a".into()), &TableIdent::new(ns.clone(), "b".into())).await.unwrap_err(),
grpc.drop_namespace(&ns).await.unwrap_err(),
grpc.purge_table(&TableIdent::new(ns.clone(), "a".into())).await.unwrap_err(),
] {
assert_eq!(e.kind(), ErrorKind::FeatureUnsupported, "{e}");
assert!(e.to_string().contains("RPC"), "refused by name: {e}");
}
}
/// `lazy` makes no call: a client for a front that is not there yet opens,
/// and its first RPC fails naming the catalog.
#[tokio::test]
async fn lazy_opens_without_a_call_and_the_first_rpc_names_the_catalog() {
let grpc = GrpcCatalog::lazy("127.0.0.1:9", Arc::new(LocalFsStorageFactory)).unwrap();
assert_eq!(grpc.uri(), "http://127.0.0.1:9");
let e = grpc.list_namespaces(None).await.unwrap_err();
assert!(e.to_string().contains("skade-katalog gRPC http://127.0.0.1:9 ListNamespaces"), "{e}");
assert!(GrpcCatalog::connect("127.0.0.1:9", Arc::new(LocalFsStorageFactory)).await.is_err());
}