skade-flight/tests/nessie_live.rs
//! **Live Nessie arms** — Power BI's Flight SQL shape against a real Nessie
//! 0.105.3 (ROCKSDB) and a real S3 (RustFS), no Spark.
//!
//! - **Arm 1 — Nessie API v2** (`skade::connect::NessieV2`, feature `nessie`).
//! Nessie runs with NO `nessie.catalog.*` config, as a Spark-fed production
//! Nessie does. The table is written by skade to `s3a://…` and registered in
//! Nessie with a v2 commit (`ICEBERG_TABLE` Put carrying `metadataLocation`).
//! - **Arm 2 — Iceberg REST** (`skade::connect::IcebergRest`, feature
//! `rest-catalog`). Nessie runs WITH its warehouse + S3 config; the table is
//! created and appended THROUGH the REST catalog (`s3://…`).
//!
//! Both serve through `Engine::with_connector` and assert: GetTables lists the
//! table, SELECT returns the rows, INSERT is refused read-only.
//!
//! Containers come up through bench's zero-shell manager (`#[path]`-included).
//! Ignored by default: `cargo test --features nessie,rest-catalog --test nessie_live -- --ignored --nocapture`.
#![cfg(any(feature = "nessie", feature = "rest-catalog"))]
use std::sync::Arc;
use std::time::Duration;
use arrow_array::{Float64Array, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use futures::TryStreamExt;
use skade::connect::CatalogConnector;
use skade::object_store::S3Config;
use skade_flight::arrow_flight::sql::CommandGetTables;
use skade_flight::arrow_flight::sql::client::FlightSqlServiceClient;
use skade_flight::engine::EngineOptions;
use skade_flight::{Auth, Engine};
use tonic::transport::Channel;
#[path = "../../bench/src/containers.rs"]
#[allow(dead_code)]
mod containers;
const BUCKET: &str = "warehouse";
fn rustfs_endpoint() -> String {
format!("http://localhost:{}", containers::rustfs_port())
}
fn s3() -> S3Config {
S3Config {
endpoint: rustfs_endpoint(),
region: "us-east-1".into(),
bucket: BUCKET.into(),
access_key_id: "rustfsadmin".into(),
secret_access_key: "rustfsadmin".into(),
path_style: true,
tls: None,
}
}
/// A per-run suffix: Nessie's RocksDB outlives a test run while its container does.
fn run_id() -> String {
let n = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis();
format!("{n}")
}
fn people() -> (Schema, RecordBatch) {
let schema = Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
Field::new("score", DataType::Float64, true),
]);
let batch = RecordBatch::try_new(
Arc::new(schema.clone()),
vec![
Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])),
Arc::new(StringArray::from(vec!["ada", "bob", "cyd", "dee", "eve"])),
Arc::new(Float64Array::from(vec![Some(9.5), Some(3.0), None, Some(7.25), Some(1.0)])),
],
)
.unwrap();
(schema, batch)
}
/// Bring a service up unless it already answers. The manager's `*_up` is blocking
/// and drives its own engine runtime, so it runs off the test's runtime.
async fn ensure_up(probe: String, up: fn() -> anyhow::Result<()>) {
let answers = {
let probe = probe.clone();
tokio::task::spawn_blocking(move || {
ureq::get(&probe).config().timeout_global(Some(Duration::from_secs(2))).build().call().is_ok()
})
.await
.unwrap()
};
if !answers {
tokio::task::spawn_blocking(up).await.unwrap().unwrap_or_else(|e| panic!("bring up {probe}: {e:#}"));
}
}
async fn ensure_bucket() {
let s = s3();
let cfg = skade_katalog::s3_storage::S3Cfg {
endpoint: s.endpoint,
region: s.region,
bucket: s.bucket,
access_key_id: s.access_key_id,
secret_access_key: s.secret_access_key,
path_style: true,
};
// Idempotent in intent; an "already exists" answer is not a failure here.
if let Err(e) = skade_katalog::s3_storage::ensure_bucket(&cfg).await {
eprintln!("ensure_bucket: {e} (continuing — the bucket may already exist)");
}
}
struct Served {
client: FlightSqlServiceClient<Channel>,
engine: Arc<Engine>,
_server: tokio::task::JoinHandle<()>,
}
async fn serve(connector: Arc<dyn CatalogConnector>, namespace: &str) -> Served {
eprintln!("connector: {} — {}", connector.kind(), connector.describe());
let opts = EngineOptions {
refresh_after: Duration::from_millis(50),
target_partitions: 2,
default_namespace: namespace.to_string(),
..Default::default()
};
let engine = Arc::new(
Engine::with_connector(Arc::clone(&connector), opts)
.await
.unwrap_or_else(|e| panic!("Engine::with_connector({}): {e}", connector.kind())),
);
assert!(engine.options().read_only, "a {} connector must make the engine read-only", connector.kind());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = {
let engine = engine.clone();
tokio::spawn(async move {
skade_flight::serve_on(engine, Arc::new(Auth::open()), listener, Default::default(), futures::future::pending())
.await
.unwrap();
})
};
let channel = Channel::from_shared(format!("http://{addr}")).unwrap().connect().await.unwrap();
Served { client: FlightSqlServiceClient::new(channel), engine, _server: server }
}
async fn fetch(c: &mut FlightSqlServiceClient<Channel>, sql: &str) -> Vec<RecordBatch> {
let info = c.execute(sql.to_string(), None).await.unwrap_or_else(|e| panic!("{sql}: {e}"));
let ticket = info.endpoint[0].ticket.clone().unwrap();
c.do_get(ticket).await.unwrap().try_collect().await.unwrap_or_else(|e| panic!("{sql}: {e}"))
}
fn rows(b: &[RecordBatch]) -> usize {
b.iter().map(|b| b.num_rows()).sum()
}
/// The Power BI shape every arm must pass: GetTables lists it, SELECT returns
/// the rows, INSERT is refused read-only.
async fn assert_power_bi_shape(served: &mut Served, namespace: &str, table: &str) {
let c = &mut served.client;
let info = c
.get_tables(CommandGetTables {
db_schema_filter_pattern: Some(namespace.to_string()),
include_schema: true,
..Default::default()
})
.await
.unwrap();
let b: Vec<RecordBatch> = c.do_get(info.endpoint[0].ticket.clone().unwrap()).await.unwrap().try_collect().await.unwrap();
let mut listed = Vec::new();
for batch in &b {
let names = batch.column_by_name("table_name").unwrap().as_any().downcast_ref::<StringArray>().unwrap();
listed.extend(names.iter().flatten().map(String::from));
}
eprintln!("GetTables({namespace}) → {listed:?}");
assert!(listed.iter().any(|n| n == table), "GetTables({namespace}) lists {listed:?}, not {table}");
let b = fetch(c, &format!("SELECT * FROM {table} ORDER BY id")).await;
eprintln!("SELECT * FROM {table} → {} row(s)", rows(&b));
assert_eq!(rows(&b), 5);
let names = b[0].column_by_name("name").unwrap().as_any().downcast_ref::<StringArray>().unwrap();
assert_eq!(names.value(0), "ada");
let b = fetch(c, &format!("SELECT count(*) AS n, sum(id) AS s FROM {table}")).await;
let n = b[0].column(0).as_any().downcast_ref::<Int64Array>().unwrap().value(0);
let s = b[0].column(1).as_any().downcast_ref::<Int64Array>().unwrap().value(0);
eprintln!("count/sum → {n}/{s}");
assert_eq!((n, s), (5, 15));
let err = c
.execute_update(format!("INSERT INTO {table} VALUES (9, 'zed', 1.0)"), None)
.await
.expect_err("INSERT must be refused on a read-only server");
eprintln!("INSERT → refused: {err}");
assert!(err.to_string().contains("read-only"), "{err}");
assert!(served.engine.tables().await.unwrap().iter().any(|t| t.name == table));
}
// ── Arm 1 — Nessie API v2, no catalog config, s3a:// ───────────────────────
#[cfg(feature = "nessie")]
const NESSIE_V2: &str = "http://localhost:19121/api/v2";
/// Write a V2 Iceberg table to `s3a://warehouse/<prefix>` through a throwaway
/// local skade catalog; return its metadata location and ids.
#[cfg(feature = "nessie")]
async fn seed_s3a_table(namespace: &str, table: &str) -> serde_json::Value {
let dir = tempfile::tempdir().unwrap();
let wh = skade::Warehouse::open_with_object_store_config(
dir.path().join("catalog.redb"),
format!("s3a://{BUCKET}/nessie-v2-{}", run_id()),
// The aegir store: what the `nessie` feature brings, and what NessieV2 reads through.
skade::object_store::ObjectStoreConfig::Aegir(s3()),
)
.await
.unwrap()
.with_new_table_format(skade::iceberg::spec::FormatVersion::V2);
let (schema, batch) = people();
let mut t = wh.create_table(&format!("{namespace}.{table}"), &schema).await.unwrap();
t.append(&[batch]).await.unwrap();
let ice = t.inner();
let md = ice.metadata();
let location = ice.metadata_location().expect("a committed table has a metadata location").to_string();
assert!(location.starts_with("s3a://"), "seeded metadata must sit at s3a:// like Spark's: {location}");
serde_json::json!({
"type": "ICEBERG_TABLE",
"metadataLocation": location,
"snapshotId": md.current_snapshot_id().unwrap_or(-1),
"schemaId": md.current_schema_id(),
"specId": md.default_partition_spec_id(),
"sortOrderId": md.default_sort_order_id(),
})
}
/// Register a namespace + table in Nessie with ONE v2 commit on `main`.
#[cfg(feature = "nessie")]
fn nessie_v2_register(namespace: &str, table: &str, content: serde_json::Value) {
let head: serde_json::Value = ureq::get(&format!("{NESSIE_V2}/trees/main")).call().unwrap().into_body().read_json().unwrap();
let hash = head["reference"]["hash"].as_str().unwrap_or_else(|| panic!("no hash in {head}"));
let body = serde_json::json!({
"commitMeta": { "message": format!("skade-flight live test: register {namespace}.{table}"), "authors": ["skade-flight-test"] },
"operations": [
{ "type": "PUT", "key": { "elements": [namespace] }, "content": { "type": "NAMESPACE", "elements": [namespace] } },
{ "type": "PUT", "key": { "elements": [namespace, table] }, "content": content },
]
});
// ureq 3: read a 4xx/5xx as a response, so the panic carries Nessie's message.
let r = ureq::post(&format!("{NESSIE_V2}/trees/main@{hash}/history/commit"))
.config()
.http_status_as_error(false)
.build()
.send_json(body)
.unwrap_or_else(|e| panic!("nessie v2 commit: {e}"));
let code = r.status().as_u16();
let text = r.into_body().read_to_string().unwrap_or_default();
if code >= 400 {
panic!("nessie v2 commit {code}: {text}");
}
eprintln!("nessie v2 commit → {text}");
}
#[cfg(feature = "nessie")]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "requires: podman socket; pulls ghcr.io/projectnessie/nessie:0.105.3 + rustfs — run with --ignored"]
async fn arm1_nessie_v2_without_catalog_config_serves_an_s3a_table() {
ensure_up(format!("{}/", rustfs_endpoint()), containers::rustfs_up).await;
ensure_up(format!("{NESSIE_V2}/config"), containers::nessie_plain_up).await;
ensure_bucket().await;
let namespace = format!("lake_{}", run_id());
let table = "people";
let content = seed_s3a_table(&namespace, table).await;
eprintln!("seeded {namespace}.{table}: {content}");
{
let (ns, content) = (namespace.clone(), content.clone());
tokio::task::spawn_blocking(move || nessie_v2_register(&ns, table, content)).await.unwrap();
}
let connector: Arc<dyn CatalogConnector> = Arc::new(skade::connect::NessieV2 {
uri: NESSIE_V2.into(),
reference: "main".into(),
token: None,
tls: None,
s3: s3(),
});
let mut served = serve(connector, &namespace).await;
assert_power_bi_shape(&mut served, &namespace, table).await;
}
// ── Arm 2 — Iceberg REST, Nessie WITH warehouse + S3 config, s3:// ─────────
#[cfg(feature = "rest-catalog")]
const NESSIE_REST: &str = "http://localhost:19120/iceberg";
#[cfg(feature = "rest-catalog")]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "requires: podman socket; pulls ghcr.io/projectnessie/nessie:0.105.3 + rustfs — run with --ignored"]
async fn arm2_iceberg_rest_with_catalog_config_serves_a_table_written_through_it() {
use iceberg::{NamespaceIdent, TableCreation};
ensure_up(format!("{}/", rustfs_endpoint()), containers::rustfs_up).await;
ensure_up(format!("{NESSIE_REST}/v1/config"), containers::nessie_up).await;
ensure_bucket().await;
let config: serde_json::Value = tokio::task::spawn_blocking(|| {
ureq::get(&format!("{NESSIE_REST}/v1/config?warehouse={BUCKET}")).call().unwrap().into_body().read_json().unwrap()
})
.await
.unwrap();
eprintln!("GET /iceberg/v1/config → {config}");
// Nessie puts the prefix in `defaults` (URL-encoded `{ref}|{warehouse}`), not `overrides`.
eprintln!(
"REST prefix returned: defaults={:?} overrides={:?} pattern={:?}",
config["defaults"]["prefix"], config["overrides"]["prefix"], config["overrides"]["nessie.prefix-pattern"]
);
let connector = skade::connect::IcebergRest {
uri: NESSIE_REST.into(),
warehouse: Some(BUCKET.into()),
prefix_or_ref: None,
token: None,
tls: None,
s3: s3(),
read_only: true,
sigv4: None,
..Default::default()
};
// Seed THROUGH the REST catalog: namespace, table, one append.
let namespace = format!("lake_{}", run_id());
let table = "people";
{
// The writer is the same catalog with read_only off; the connector served
// below stays read-only, and refuses exactly these calls.
let writer = skade::connect::IcebergRest {
uri: NESSIE_REST.into(),
warehouse: Some(BUCKET.into()),
prefix_or_ref: None,
token: None,
tls: None,
s3: s3(),
read_only: false,
sigv4: None,
..Default::default()
};
let catalog = writer.open().await.unwrap();
let ns = NamespaceIdent::new(namespace.clone());
catalog.create_namespace(&ns, Default::default()).await.unwrap();
let (schema, batch) = people();
let creation = TableCreation::builder()
.name(table.to_string())
.schema(skade::arrow_to_iceberg(&schema).unwrap())
.build();
let created = catalog.create_table(&ns, creation).await.unwrap();
let appended = skade::append(catalog.as_ref(), &created, &[batch]).await.unwrap();
eprintln!("seeded {namespace}.{table} at {:?}", appended.metadata_location());
}
let mut served = serve(Arc::new(connector), &namespace).await;
assert_power_bi_shape(&mut served, &namespace, table).await;
}