diff --git a/crates/ourios-parquet/src/lib.rs b/crates/ourios-parquet/src/lib.rs index 7ed7300a..2ea3cc5f 100644 --- a/crates/ourios-parquet/src/lib.rs +++ b/crates/ourios-parquet/src/lib.rs @@ -28,6 +28,7 @@ pub mod audit_reader; pub mod audit_record_batch; pub mod audit_writer; +pub mod manifest; pub mod partition; pub mod reader; pub mod record_batch; @@ -36,6 +37,7 @@ pub mod writer; pub use audit_reader::{AuditReader, AuditReaderError}; pub use audit_record_batch::{AuditBatchError, audit_events_to_batch}; pub use audit_writer::{AuditWriter, AuditWriterError, AuditWrittenFile}; +pub use manifest::{MANIFEST_FILENAME, Manifest, ManifestError}; pub use partition::{PartitionKey, TimestampOverflowError, percent_encode_tenant}; pub use reader::{Reader, ReaderError}; pub use record_batch::{BatchError, mined_records_to_batch}; diff --git a/crates/ourios-parquet/src/manifest.rs b/crates/ourios-parquet/src/manifest.rs new file mode 100644 index 00000000..9606501a --- /dev/null +++ b/crates/ourios-parquet/src/manifest.rs @@ -0,0 +1,272 @@ +//! Per-partition manifest (RFC 0009 §3.4). +//! +//! Compaction can't atomically replace a partition's many small +//! `*.parquet` files with one consolidated file on object storage +//! (no atomic multi-object operation), so a directory glob would race +//! a compaction and double-count or miss rows. The **manifest** names +//! the authoritative live set of data files in a partition plus a +//! monotonically increasing generation; the read path (RFC 0007) +//! resolves a partition's files through it, and a compaction commits +//! by atomically swapping the manifest. A partition with no manifest +//! — every partition today, pre-compaction — falls back to "all +//! committed `*.parquet`", so the manifest is additive and +//! back-compatible (RFC 0009 §3.4 reader-first sequencing: the reader +//! learns the manifest before any compactor writes one). + +use std::path::Path; + +use serde::{Deserialize, Serialize}; + +/// Canonical manifest filename inside a partition directory. +pub const MANIFEST_FILENAME: &str = "manifest.json"; + +/// The live data-file set of one partition (RFC 0009 §3.4). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Manifest { + /// Monotonic generation, bumped on each atomic swap. Lets a reader + /// confirm it read one consistent generation and a compactor + /// detect a lost update. + pub generation: u64, + /// Live data files, as bare file names (no path) relative to the + /// partition directory. A reader joins each onto the partition + /// dir; files present on disk but *not* listed here are orphans + /// awaiting GC and MUST be ignored. + pub files: Vec, +} + +/// Failure reading, parsing, or validating a [`Manifest`]. +#[derive(Debug)] +#[non_exhaustive] +pub enum ManifestError { + /// The manifest file existed but could not be read. + Io(std::io::Error), + /// The manifest bytes were not valid manifest JSON. + Parse(serde_json::Error), + /// A `files` entry is not a partition-local `*.parquet` file name + /// (it is absolute, contains path separators, or escapes the + /// partition directory via `..`). A reader joins entries onto the + /// partition dir, so accepting such a name would let a manifest + /// point a query at files outside the tenant's partition — + /// breaking tenant isolation (`CLAUDE.md` §3.7 / RFC0007.5). Bad + /// manifests fail loudly rather than silently mis-resolving. + InvalidFilename(String), +} + +impl std::fmt::Display for ManifestError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Io(e) => write!(f, "read manifest: {e}"), + Self::Parse(e) => write!(f, "parse manifest: {e}"), + Self::InvalidFilename(name) => { + write!( + f, + "manifest lists a non-partition-local file name: {name:?}" + ) + } + } + } +} + +impl std::error::Error for ManifestError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io(e) => Some(e), + Self::Parse(e) => Some(e), + Self::InvalidFilename(_) => None, + } + } +} + +/// Whether `name` is a bare partition-local `*.parquet` file name: +/// exactly one path component, that component an ordinary name (no +/// `/`, no `..`, not absolute), with a lowercase `.parquet` extension. +/// The extension match is case-*sensitive* on purpose — the writer +/// only ever emits lowercase `.parquet`, and the glob fallback +/// (`resolve_live_files`) / `ListingOptions::with_file_extension` +/// match lowercase too, so a manifest naming `*.PARQUET` would be +/// "valid" yet inconsistent with the on-disk contract. +fn is_partition_local_parquet(name: &str) -> bool { + use std::path::Component; + let path = Path::new(name); + let mut components = path.components(); + let single_normal = + matches!(components.next(), Some(Component::Normal(_))) && components.next().is_none(); + single_normal && path.extension().is_some_and(|ext| ext == "parquet") +} + +impl Manifest { + /// Read `/manifest.json`. + /// + /// `Ok(None)` when the manifest is absent — the pre-compaction + /// (and current) case, where the reader falls back to globbing + /// `*.parquet`. `Ok(Some(_))` when present and parseable. + /// + /// # Errors + /// + /// [`ManifestError`] if the file exists but can't be read, or its + /// bytes aren't valid manifest JSON. + pub fn read(partition_dir: &Path) -> Result, ManifestError> { + match std::fs::read(partition_dir.join(MANIFEST_FILENAME)) { + Ok(bytes) => { + let manifest: Self = + serde_json::from_slice(&bytes).map_err(ManifestError::Parse)?; + manifest.validate()?; + Ok(Some(manifest)) + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(e) => Err(ManifestError::Io(e)), + } + } + + /// Reject any `files` entry that is not a partition-local + /// `*.parquet` file name (see [`ManifestError::InvalidFilename`]). + /// + /// # Errors + /// + /// [`ManifestError::InvalidFilename`] for the first offending name. + pub fn validate(&self) -> Result<(), ManifestError> { + for name in &self.files { + if !is_partition_local_parquet(name) { + return Err(ManifestError::InvalidFilename(name.clone())); + } + } + Ok(()) + } + + /// Serialize to the canonical JSON bytes the compactor writes and + /// [`read`](Self::read) parses. + /// + /// # Errors + /// + /// [`serde_json::Error`] if serialization fails (not expected for + /// this plain struct). + pub fn to_json(&self) -> Result, serde_json::Error> { + serde_json::to_vec(self) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn round_trips_through_json() { + // Arrange + let manifest = Manifest { + generation: 7, + files: vec!["a.parquet".to_string(), "b.parquet".to_string()], + }; + + // Act + let restored: Manifest = + serde_json::from_slice(&manifest.to_json().expect("serialize")).expect("parse"); + + // Assert + assert_eq!(restored, manifest); + } + + #[test] + fn absent_manifest_is_none() { + // Arrange + let dir = tempfile::tempdir().expect("temp"); + + // Act + let read = Manifest::read(dir.path()); + + // Assert + assert_eq!(read.expect("read"), None); + } + + #[test] + fn reads_a_written_manifest() { + // Arrange + let dir = tempfile::tempdir().expect("temp"); + let manifest = Manifest { + generation: 3, + files: vec!["compacted.parquet".to_string()], + }; + std::fs::write( + dir.path().join(MANIFEST_FILENAME), + manifest.to_json().unwrap(), + ) + .expect("write"); + + // Act + let read = Manifest::read(dir.path()); + + // Assert + assert_eq!(read.expect("read"), Some(manifest)); + } + + #[test] + fn malformed_manifest_is_a_parse_error() { + // Arrange + let dir = tempfile::tempdir().expect("temp"); + std::fs::write(dir.path().join(MANIFEST_FILENAME), b"not json").expect("write"); + + // Act + let read = Manifest::read(dir.path()); + + // Assert + assert!(matches!(read, Err(ManifestError::Parse(_)))); + } + + #[test] + fn accepts_plain_parquet_names() { + // Arrange — bare partition-local names a compactor would emit. + let names = ["a.parquet", "01890000-0000-7000-8000-000000000000.parquet"]; + + // Act & Assert (table-driven over the pure predicate) + for name in names { + assert!( + is_partition_local_parquet(name), + "{name} should be accepted" + ); + } + } + + #[test] + fn rejects_path_escaping_or_non_parquet_names() { + // Arrange — names that escape the partition or aren't parquet. + let names = [ + "../escape.parquet", // parent escape + "/abs/x.parquet", // absolute + "sub/x.parquet", // nested + "x.txt", // wrong extension + "x.PARQUET", // non-canonical (uppercase) extension + "x", // no extension + "", // empty + ".", // current dir + ]; + + // Act & Assert (table-driven over the pure predicate) + for name in names { + assert!( + !is_partition_local_parquet(name), + "{name:?} should be rejected" + ); + } + } + + #[test] + fn read_rejects_a_path_escaping_entry() { + // Arrange — a hostile manifest on disk (serialized directly, + // bypassing `validate`) whose entry escapes the partition. + let dir = tempfile::tempdir().expect("temp"); + let evil = Manifest { + generation: 1, + files: vec!["../../../etc/secrets.parquet".to_string()], + }; + std::fs::write( + dir.path().join(MANIFEST_FILENAME), + serde_json::to_vec(&evil).unwrap(), + ) + .expect("write"); + + // Act + let read = Manifest::read(dir.path()); + + // Assert + assert!(matches!(read, Err(ManifestError::InvalidFilename(_)))); + } +} diff --git a/crates/ourios-querier/src/lib.rs b/crates/ourios-querier/src/lib.rs index 3fd58089..2ae7a561 100644 --- a/crates/ourios-querier/src/lib.rs +++ b/crates/ourios-querier/src/lib.rs @@ -37,7 +37,6 @@ use std::path::PathBuf; use std::sync::Arc; use datafusion::arrow::array::{Array, Int64Array}; -use datafusion::arrow::datatypes::DataType; use datafusion::arrow::record_batch::RecordBatch; use datafusion::common::ScalarValue; use datafusion::datasource::file_format::parquet::ParquetFormat; @@ -50,6 +49,7 @@ use datafusion::physical_plan::metrics::{MetricValue, MetricsSet}; use datafusion::physical_plan::{ExecutionPlan, collect}; use datafusion::prelude::{SessionContext, col, lit}; use ourios_core::tenant::TenantId; +use ourios_parquet::Manifest; use ourios_parquet::columns; use ourios_parquet::percent_encode_tenant; @@ -143,44 +143,61 @@ impl std::fmt::Display for QueryError { impl std::error::Error for QueryError {} -/// Whether `dir` (a tenant's partition root) holds at least one -/// published `*.parquet` file anywhere beneath it. Recursive -/// because the data is nested `year=/month=/day=/hour=/`. Files -/// the writer hasn't committed (`*.parquet.tmp`) have extension -/// `tmp`, so they don't count — the poisoned-writer case we treat -/// as "empty", not error. +/// Resolve the live data files a query must read under `dir` (a +/// tenant's partition root), honouring the RFC 0009 §3.4 +/// per-partition manifest. Recursive because the data is nested +/// `year=/month=/day=/hour=/`. /// -/// A missing directory (`NotFound`) is "empty" (`Ok(false)`); any -/// *other* I/O error (permission denied, transient failure) is -/// propagated as [`QueryError::Storage`] rather than silently -/// masked as "no data" — a wrong zero-row answer is worse than a -/// surfaced error. -fn has_published_parquet(dir: &std::path::Path) -> Result { +/// For each partition directory: if it holds a `manifest.json`, the +/// manifest is authoritative and contributes exactly the files it +/// names (files present on disk but not listed — orphans awaiting GC, +/// or a writer's uncommitted `*.parquet.tmp` — are ignored). With no +/// manifest (every partition today, pre-compaction) it falls back to +/// all committed `*.parquet` in that directory; `*.parquet.tmp` has +/// extension `tmp`, so the poisoned-writer case contributes nothing. +/// +/// An empty result means the tenant has nothing queryable. A missing +/// directory (`NotFound`) is empty; any *other* I/O error (permission +/// denied, transient failure) is propagated as [`QueryError::Storage`] +/// rather than silently masked as "no data" — a wrong zero-row answer +/// is worse than a surfaced error. +fn resolve_live_files(dir: &std::path::Path) -> Result, QueryError> { let io_err = |op: &str, p: &std::path::Path, e: &std::io::Error| QueryError::Storage { detail: format!("{op} {}: {e}", p.display()), }; + let mut files = Vec::new(); let mut stack = vec![dir.to_path_buf()]; while let Some(d) = stack.pop() { let entries = match std::fs::read_dir(&d) { Ok(entries) => entries, - // The dir (or a subdir, lost to a concurrent - // housekeeping unlink) simply isn't there → not data, - // not an error. + // The dir (or a subdir, lost to a concurrent housekeeping + // unlink) simply isn't there → not data, not an error. Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue, Err(e) => return Err(io_err("read_dir", &d, &e)), }; + let mut subdirs = Vec::new(); + let mut parquets = Vec::new(); for entry in entries { let entry = entry.map_err(|e| io_err("read_dir entry", &d, &e))?; let path = entry.path(); match entry.file_type() { - Ok(ft) if ft.is_dir() => stack.push(path), - Ok(_) if path.extension().is_some_and(|x| x == "parquet") => return Ok(true), + Ok(ft) if ft.is_dir() => subdirs.push(path), + Ok(_) if path.extension().is_some_and(|x| x == "parquet") => parquets.push(path), Ok(_) => {} Err(e) => return Err(io_err("file_type", &path, &e)), } } + match Manifest::read(&d).map_err(|e| QueryError::Storage { + detail: format!("manifest in {}: {e}", d.display()), + })? { + // Manifest is authoritative: only its named files are live. + Some(manifest) => files.extend(manifest.files.into_iter().map(|name| d.join(name))), + // No manifest → glob fallback for this partition. + None => files.append(&mut parquets), + } + stack.extend(subdirs); } - Ok(false) + Ok(files) } /// Pull the single aggregate count out of the result batches. A @@ -312,48 +329,70 @@ impl Querier { .bucket_root .join("data") .join(format!("tenant_id={enc}")); - // No published `*.parquet` under the tenant dir ⇒ the - // tenant has nothing queryable ⇒ empty result (not an - // error). Covers both the missing-dir case and a dir that - // holds only `*.parquet.tmp` (a poisoned/crashed writer) or - // empty partition dirs — where `infer_schema` would - // otherwise error and wrongly fail the query. - if !has_published_parquet(&tenant_dir)? { + // Resolve the live file set under the tenant dir, honouring + // the RFC 0009 §3.4 manifest (glob-fallback when absent). An + // empty set ⇒ the tenant has nothing queryable ⇒ empty result + // (not an error). Covers the missing-dir case and a partition + // holding only `*.parquet.tmp` (a poisoned/crashed writer) — + // where building a table over zero files would otherwise + // error and wrongly fail the query. + let live_files = resolve_live_files(&tenant_dir)?; + if live_files.is_empty() { return Ok(QueryResult::default()); } let ctx = SessionContext::new(); - // Build the table URL from the canonical absolute path, - // scheme-less with a trailing slash. DataFusion 53 treats - // an absolute filesystem path as local and URI-encodes it - // internally — so spaces / reserved characters in the - // bucket path are handled, unlike a hand-built `file://…` - // string. `canonicalize` is safe: we just confirmed the - // directory exists. The trailing slash marks it a - // directory (not a single object). - let abs = tenant_dir.canonicalize().map_err(|e| QueryError::Storage { + // Tenant isolation (RFC0007.5 / §3.7) is enforced here, not + // just assumed structural: every resolved file must + // canonicalize to a path *under* the tenant's canonical + // partition root. The manifest's entries are already validated + // as partition-local names (`Manifest::validate`), but a + // symlinked `*.parquet` could still resolve outside — this + // `starts_with` check is the backstop that fails such a path + // loudly rather than reading another tenant's data. + let tenant_root = tenant_dir.canonicalize().map_err(|e| QueryError::Storage { detail: format!("canonicalize {}: {e}", tenant_dir.display()), })?; - let url = ListingTableUrl::parse(format!("{}/", abs.display())).map_err(storage_err)?; - // `year/month/day/hour` are path-only Hive partition cols - // (parsed from the directory names); `tenant_id` is *not* - // listed — relative to this tenant-scoped root it's a plain - // file column, and the rooting is what enforces isolation. - let options = ListingOptions::new(Arc::new(ParquetFormat::default())) - .with_file_extension(".parquet") - .with_table_partition_cols(vec![ - ("year".to_string(), DataType::Utf8), - ("month".to_string(), DataType::Utf8), - ("day".to_string(), DataType::Utf8), - ("hour".to_string(), DataType::Utf8), - ]); - let schema = options - .infer_schema(&ctx.state(), &url) + // One table path per *live* data file (RFC 0009 §3.4 — the + // manifest, not a directory glob, decides the file set, so a + // query never sees a compaction's superseded inputs). Each is + // the canonical absolute path: DataFusion 53 treats an + // absolute filesystem path as local and URI-encodes it + // internally, so spaces / reserved characters are handled + // without a hand-built `file://…` string. `year/month/day/hour` + // stay path-only (not file columns) and the query filters only + // data columns, so no table partition columns are declared. + let mut seen = std::collections::HashSet::new(); + let mut urls = Vec::with_capacity(live_files.len()); + for file in &live_files { + let abs = file.canonicalize().map_err(|e| QueryError::Storage { + detail: format!("canonicalize {}: {e}", file.display()), + })?; + if !abs.starts_with(&tenant_root) { + return Err(QueryError::Storage { + detail: format!( + "resolved file {} escapes tenant partition root {}", + abs.display(), + tenant_root.display(), + ), + }); + } + // De-duplicate so a manifest naming the same file twice + // can't double-count its rows. + if seen.insert(abs.clone()) { + urls.push(ListingTableUrl::parse(abs.display().to_string()).map_err(storage_err)?); + } + } + let options = + ListingOptions::new(Arc::new(ParquetFormat::default())).with_file_extension(".parquet"); + // `infer_schema` over the multi-path set merges the files' + // schemas, so additive schema drift across files reads as the + // union (RFC0007.4 / RFC 0005 §3.9). + let config = ListingTableConfig::new_with_multi_paths(urls) + .with_listing_options(options) + .infer_schema(&ctx.state()) .await .map_err(storage_err)?; - let config = ListingTableConfig::new(url) - .with_listing_options(options) - .with_schema(schema); let table = ListingTable::try_new(config).map_err(storage_err)?; ctx.register_table("logs", Arc::new(table)) .map_err(storage_err)?; @@ -545,4 +584,105 @@ mod tests { assert_eq!(stats.row_groups_scanned, 2); assert_eq!(stats.bytes_read, 4096); } + + // --- resolve_live_files (RFC 0009 §3.4 manifest / glob fallback) --- + + /// Create `/data/tenant_id=a/year=2026/.../hour=10` and + /// return `(tenant_dir, partition_dir)`. + fn tenant_and_partition(root: &std::path::Path) -> (PathBuf, PathBuf) { + let tenant = root.join("data/tenant_id=a"); + let partition = tenant.join("year=2026/month=04/day=02/hour=10"); + std::fs::create_dir_all(&partition).expect("mkdir partition"); + (tenant, partition) + } + + #[test] + fn resolve_missing_tenant_dir_is_empty() { + // Arrange — a tenant directory that was never written. + let tmp = tempfile::tempdir().expect("temp"); + let ghost = tmp.path().join("data/tenant_id=ghost"); + + // Act + let files = resolve_live_files(&ghost).expect("resolve"); + + // Assert + assert!(files.is_empty()); + } + + #[test] + fn resolve_tmp_only_partition_is_empty() { + // Arrange — a partition holding only an uncommitted `.tmp`. + let tmp = tempfile::tempdir().expect("temp"); + let (tenant, partition) = tenant_and_partition(tmp.path()); + std::fs::write(partition.join("x.parquet.tmp"), b"partial").expect("write tmp"); + + // Act + let files = resolve_live_files(&tenant).expect("resolve"); + + // Assert + assert!(files.is_empty(), "uncommitted .tmp files are not live"); + } + + #[test] + fn resolve_globs_committed_parquet_without_a_manifest() { + // Arrange — two committed files, no manifest. + let tmp = tempfile::tempdir().expect("temp"); + let (tenant, partition) = tenant_and_partition(tmp.path()); + std::fs::write(partition.join("a.parquet"), b"a").expect("write a"); + std::fs::write(partition.join("b.parquet"), b"b").expect("write b"); + + // Act + let files = resolve_live_files(&tenant).expect("resolve"); + + // Assert + assert_eq!( + files.len(), + 2, + "both committed files are live without a manifest" + ); + } + + #[test] + fn resolve_manifest_is_authoritative() { + // Arrange — two files on disk, a manifest naming only one. + let tmp = tempfile::tempdir().expect("temp"); + let (tenant, partition) = tenant_and_partition(tmp.path()); + std::fs::write(partition.join("a.parquet"), b"a").expect("write a"); + std::fs::write(partition.join("b.parquet"), b"b").expect("write b"); + let manifest = ourios_parquet::Manifest { + generation: 1, + files: vec!["a.parquet".to_string()], + }; + std::fs::write( + partition.join(ourios_parquet::MANIFEST_FILENAME), + manifest.to_json().unwrap(), + ) + .expect("write manifest"); + + // Act + let files = resolve_live_files(&tenant).expect("resolve"); + + // Assert + assert_eq!(files.len(), 1, "only the manifest's file is live"); + assert!(files[0].ends_with("a.parquet")); + } + + #[test] + fn resolve_malformed_manifest_is_a_storage_error() { + // Arrange — a manifest that isn't valid JSON. + let tmp = tempfile::tempdir().expect("temp"); + let (tenant, partition) = tenant_and_partition(tmp.path()); + std::fs::write(partition.join("a.parquet"), b"a").expect("write a"); + std::fs::write( + partition.join(ourios_parquet::MANIFEST_FILENAME), + b"not json", + ) + .expect("write manifest"); + + // Act + let result = resolve_live_files(&tenant); + + // Assert + assert!(matches!(result, Err(QueryError::Storage { .. }))); + } } diff --git a/crates/ourios-querier/tests/manifest.rs b/crates/ourios-querier/tests/manifest.rs new file mode 100644 index 00000000..b65822ee --- /dev/null +++ b/crates/ourios-querier/tests/manifest.rs @@ -0,0 +1,158 @@ +//! RFC 0009 §3.4 reader side (sequenced first per RFC0009 §7): the +//! querier resolves a partition's files through a `manifest.json` +//! when present, falling back to a `*.parquet` glob when absent. This +//! is the read-half of RFC0009.3 (a query reads one consistent +//! generation — never a compaction's superseded inputs). The +//! compactor that *writes* manifests is a later slice (epic #94); a +//! manifest is hand-written here to stand in for a committed +//! compaction. + +use std::path::{Path, PathBuf}; + +use ourios_core::audit::ParamType; +use ourios_core::record::{BodyKind, MinedRecord, Param}; +use ourios_core::tenant::TenantId; +use ourios_parquet::{MANIFEST_FILENAME, Manifest, PartitionKey, Writer}; +use ourios_querier::{Querier, QueryRequest}; + +/// 2026-04-02T10:58:00 UTC — all offsets below stay within hour 10, +/// so every file lands in the same partition directory. +const TS0: u64 = 1_775_127_480_000_000_000; + +fn rec(template_id: u64, ts_ns: u64) -> MinedRecord { + MinedRecord { + tenant_id: TenantId::new("a"), + template_id, + template_version: 1, + severity_number: 9, + severity_text: Some("INFO".to_string()), + scope_name: Some("lib.cart".to_string()), + scope_version: Some("1.0.0".to_string()), + time_unix_nano: ts_ns, + observed_time_unix_nano: Some(ts_ns + 1_000), + attributes: Vec::new(), + dropped_attributes_count: 0, + resource_attributes: Vec::new(), + trace_id: None, + span_id: None, + flags: 0x01, + event_name: None, + body_kind: BodyKind::String, + params: vec![Param { + type_tag: ParamType::Num, + value: "42".to_string(), + }], + separators: vec![String::new(), " ".to_string()], + body: None, + confidence: 1.0, + lossy_flag: false, + } +} + +fn req(template_id: Option) -> QueryRequest { + QueryRequest { + tenant: TenantId::new("a"), + time_range: None, + template_id, + } +} + +/// Write `recs` (which must share a partition) as one committed file; +/// return the partition directory and the committed file name. +fn write_file(bucket: &Path, recs: &[MinedRecord]) -> (PathBuf, String) { + let part = PartitionKey::derive(&recs[0]).expect("derive partition"); + let mut w = Writer::open(bucket, part).expect("open writer"); + w.append_records(recs).expect("append"); + let written = w.close().expect("close"); + let name = written + .path + .file_name() + .and_then(|s| s.to_str()) + .expect("file name") + .to_string(); + let dir = written.path.parent().expect("parent").to_path_buf(); + (dir, name) +} + +/// A manifest is authoritative: with two files in one partition but a +/// manifest naming only one, the query sees only the named file's +/// rows — the unlisted file is ignored even though it is on disk. +#[tokio::test] +async fn rfc0009_3_manifest_restricts_to_named_files() { + let bucket = tempfile::TempDir::new().expect("temp"); + let (dir_a, file_a) = write_file(bucket.path(), &[rec(1, TS0), rec(1, TS0 + 1_000_000)]); + let (dir_b, _file_b) = write_file( + bucket.path(), + &[rec(2, TS0 + 2_000_000), rec(2, TS0 + 3_000_000)], + ); + assert_eq!(dir_a, dir_b, "both files share the hour-10 partition"); + + let q = Querier::new(bucket.path()); + + // No manifest yet → glob fallback sees both files. + let all = q.run(req(None)).await.expect("glob-fallback query"); + assert_eq!(all.rows, 4, "without a manifest, both files are read"); + + // Manifest naming only file A ⇒ file B is no longer live. + let manifest = Manifest { + generation: 1, + files: vec![file_a], + }; + std::fs::write(dir_a.join(MANIFEST_FILENAME), manifest.to_json().unwrap()) + .expect("write manifest"); + + let scoped = q.run(req(None)).await.expect("manifest query"); + assert_eq!( + scoped.rows, 2, + "the manifest's single file is authoritative" + ); +} + +/// Models a committed compaction (RFC0009.3): the partition holds the +/// two original inputs *and* a consolidated file with all their rows, +/// and the manifest names only the consolidated file. The query must +/// return each row exactly once — the superseded inputs are not +/// double-counted. +#[tokio::test] +async fn rfc0009_3_manifest_naming_compacted_file_avoids_double_count() { + let bucket = tempfile::TempDir::new().expect("temp"); + write_file(bucket.path(), &[rec(1, TS0), rec(1, TS0 + 1_000_000)]); + write_file( + bucket.path(), + &[rec(2, TS0 + 2_000_000), rec(2, TS0 + 3_000_000)], + ); + // The "compacted" output: all four rows in one file. + let (dir, compacted) = write_file( + bucket.path(), + &[ + rec(1, TS0), + rec(1, TS0 + 1_000_000), + rec(2, TS0 + 2_000_000), + rec(2, TS0 + 3_000_000), + ], + ); + + let q = Querier::new(bucket.path()); + + // Pre-commit (no manifest): the glob sees inputs *and* the + // compacted file — the double-count the manifest exists to prevent. + let pre = q.run(req(None)).await.expect("pre-commit query"); + assert_eq!(pre.rows, 8, "glob without a manifest double-counts"); + + // Commit: manifest names only the compacted file. + let manifest = Manifest { + generation: 2, + files: vec![compacted], + }; + std::fs::write(dir.join(MANIFEST_FILENAME), manifest.to_json().unwrap()) + .expect("write manifest"); + + let post = q.run(req(None)).await.expect("post-commit query"); + assert_eq!( + post.rows, 4, + "after the manifest commit, each row counts once" + ); + // And template-exact pushdown still works against the manifested set. + let t1 = q.run(req(Some(1))).await.expect("template query"); + assert_eq!(t1.rows, 2); +}