bench/tests/bench_arms.rs
//! Light functional smoke for the two Phase-4 bench arms, so they exist and
//! compile and actually drive the scenarios (the HEAVY authoritative numbers
//! are Loki's job). Quick scale, in-process.
use iceberg::{Catalog, NamespaceIdent, TableIdent};
use skade_katalog_bench::grpc_catalog::GrpcCatalog;
use skade_katalog_bench::{data, factory, scenarios};
use scenarios::Scale;
fn tiny() -> Scale {
Scale { tables: 4, iters: 40, threads: vec![1, 2] }
}
/// The `skade-grpc` arm: embedded catalog behind the gRPC front, driven through
/// the `GrpcCatalog` adapter by the *same* generic scenarios as embedded/REST.
#[tokio::test]
async fn grpc_arm_runs_the_scenarios() {
let (cat, _tmp) = factory::embedded().await.unwrap();
let bound = skade_katalog_grpc::spawn(cat, "127.0.0.1:0".parse().unwrap())
.await
.unwrap();
let grpc = GrpcCatalog::connect(bound).await.unwrap();
let scale = tiny();
let ns = scenarios::seed(&grpc, &scale).await.unwrap();
// A seeded table is visible over the gRPC wire.
assert!(
grpc.table_exists(&TableIdent::new(ns.clone(), "bench_t_000000".to_string()))
.await
.unwrap(),
"seeded table exists over gRPC"
);
let exists = scenarios::table_exists_latency(&grpc, &ns, &scale).await;
assert!(exists.ops > 0, "table_exists arm produced samples");
let load = scenarios::load_table_latency(&grpc, &ns, &scale).await;
assert!(load.ops > 0, "load_table arm produced samples");
}
/// The serverless opponent arm: iceberg-rust's `MemoryCatalog` through the same
/// scenarios (the fair no-service rival for the embedded column).
#[tokio::test]
async fn memory_opponent_arm_runs_the_scenarios() {
let (cat, _tmp) = factory::memory().await.unwrap();
let scale = tiny();
let ns: NamespaceIdent = scenarios::seed(&cat, &scale).await.unwrap();
let exists = scenarios::table_exists_latency(&cat, &ns, &scale).await;
assert!(exists.ops > 0, "memory-catalog table_exists produced samples");
}
/// Testmatrix row proving the three `_s3` arms' self-skip is HONEST, not a false
/// green. When the bench is built without `--features s3` (the default `cargo
/// test` build), each S3 data-plane arm emits `factory::s3_not_compiled_skip_metrics()`
/// wrapped as a `BenchSource::Static` row. This asserts the payload:
/// 1. carries an explicit `skipped=1` flag and a reason that names the fix; and
/// 2. carries NONE of the throughput keys a real measured run emits
/// (`rows_per_sec`, `commits_per_sec`, `ops_sec`, `commits`, `files`), so no
/// consumer — README render, no-regression gate, operator — can mistake the
/// skip for a real result.
/// Red-when-broken: fabricating a throughput number into the skip (the classic
/// false-green mistake) trips assertion 2; dropping the skip flag/reason trips 1.
#[test]
fn s3_arms_skip_is_honest_not_false_green() {
let m = factory::s3_not_compiled_skip_metrics();
assert_eq!(
m.get("skipped").and_then(|v| v.as_f64()),
Some(1.0),
"skip row must carry an explicit `skipped=1` flag"
);
let reason = m.get("skip_reason").and_then(|v| v.as_str()).unwrap_or_default();
assert!(
reason.contains("--features s3"),
"skip reason must name the fix (`--features s3`); got {reason:?}"
);
for k in ["rows_per_sec", "commits_per_sec", "ops_sec", "commits", "files"] {
assert!(
!m.contains_key(k),
"skip row must not carry the measured throughput key {k:?} (that would be a false green)"
);
}
}
/// RED-when-broken guard for the ROOT-LAW #0 / reuse fix: `data::read_row_groups`
/// (and, through it, `read_parquet_parallel`) no longer hand-rolls a private
/// OS-thread bucket pool — it BINDS the shipped `skade::parquet_io` gatling
/// reader. This asserts the shared path decodes the EXACT rows expected from a
/// multi-row-group file:
/// 1. the all-core fan-out (`threads=4`) over 4 row groups returns every row
/// (order across buckets isn't preserved, so compare the sorted id set);
/// 2. it is byte-identical to a lean single-stream read (`threads=1`); and
/// 3. `read_parquet_parallel` caps to exactly `max_rows`.
/// Red-when-broken: re-introducing a divergent private pool, dropping row
/// groups, or mangling the schema trips these equalities.
#[test]
fn shared_gatling_reader_decodes_expected_rows() {
use arrow_array::{Int64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use parquet::arrow::ArrowWriter;
use parquet::file::properties::WriterProperties;
use std::sync::Arc;
// Write 400 int64 ids in 4 row groups (100 rows each).
let tmp = tempfile::tempdir().unwrap();
let path = tmp.path().join("multi.parquet");
let schema = Arc::new(ArrowSchema::new(vec![Field::new("id", DataType::Int64, false)]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![Arc::new(Int64Array::from((0..400i64).collect::<Vec<_>>()))],
)
.unwrap();
let props = WriterProperties::builder()
.set_max_row_group_row_count(Some(100))
.build();
let file = std::fs::File::create(&path).unwrap();
let mut wr = ArrowWriter::try_new(file, schema, Some(props)).unwrap();
wr.write(&batch).unwrap();
wr.close().unwrap();
let path = path.to_str().unwrap();
let ids_of = |batches: &[RecordBatch]| -> Vec<i64> {
let mut v: Vec<i64> = Vec::new();
for b in batches {
let a = b.column(0).as_any().downcast_ref::<Int64Array>().unwrap();
v.extend((0..a.len()).map(|i| a.value(i)));
}
v.sort_unstable();
v
};
let expected: Vec<i64> = (0..400).collect();
// (1) + (2): all-core fan-out equals the single stream equals the expected rows.
let fanned = data::read_row_groups(path, &[0, 1, 2, 3], 64, 4).unwrap();
let single = data::read_row_groups(path, &[0, 1, 2, 3], 64, 1).unwrap();
assert_eq!(ids_of(&single), expected, "single-stream decode must return all 400 rows");
assert_eq!(
ids_of(&fanned),
expected,
"gatling fan-out must decode the exact same rows as the single stream"
);
// (3): read_parquet_parallel caps to exactly max_rows and keeps the id set.
let (_ice, capped) = data::read_parquet_parallel(path, 250, 64, 4).unwrap();
assert_eq!(
capped.iter().map(|b| b.num_rows()).sum::<usize>(),
250,
"read_parquet_parallel must cap to exactly max_rows"
);
}