From 13ba1516d2715c0db34e8f05e9623574be00bb04 Mon Sep 17 00:00:00 2001 From: Isaac Elbaz Date: Sat, 5 Sep 2026 09:35:09 -0400 Subject: [PATCH] Fix unbounded temporal intervals and Allen relations --- src/persistence.rs | 73 ++++++++++++++++++++--- src/temporal.rs | 83 +++++++++++++++++++++----- tests/distributed_e2e.rs | 56 ++++++++++++++++++ tests/temporal_sampling.rs | 117 +++++++++++++++++++++++++++++++++++++ 4 files changed, 305 insertions(+), 24 deletions(-) create mode 100644 tests/temporal_sampling.rs diff --git a/src/persistence.rs b/src/persistence.rs index 6e89b2f..6d0eced 100644 --- a/src/persistence.rs +++ b/src/persistence.rs @@ -489,6 +489,10 @@ const KEY_DSU_CLUSTER_COUNT: &[u8] = b"dsu_cluster_count"; const STORAGE_FORMAT_VERSION: u32 = 1; const INDEX_FORMAT_VERSION: u32 = 2; const TEMPORAL_BUCKET_SECONDS: i64 = 86400; +const MAX_TEMPORAL_BUCKETS: i64 = 4096; +// No timestamp divided by TEMPORAL_BUCKET_SECONDS can produce this bucket. +// Existing day-indexed records remain readable without rebuilding their indexes. +const WIDE_TEMPORAL_BUCKET: i64 = i64::MIN; const DEFAULT_CACHE_CAPACITY: usize = 100_000; const DEFAULT_BLOCK_CACHE_MB: u64 = 512; const DEFAULT_WRITE_BUFFER_MB: u64 = 128; @@ -1598,6 +1602,9 @@ impl RecordStore for PersistentStore { } fn get_records_in_interval(&self, interval: crate::temporal::Interval) -> Vec { + if interval.is_empty() { + return Vec::new(); + } let cf = match self.db.cf_handle(CF_INDEX_TEMPORAL_BUCKET) { Some(cf) => cf, None => { @@ -1606,12 +1613,11 @@ impl RecordStore for PersistentStore { } }; let mut candidates = std::collections::HashSet::new(); - for bucket in buckets_for_interval(interval.start, interval.end) { - let prefix = bucket.to_be_bytes(); - let iter = self - .db - .iterator_cf(cf, IteratorMode::From(&prefix, Direction::Forward)); - for entry in iter { + let buckets = buckets_for_interval(interval.start, interval.end); + if buckets == [WIDE_TEMPORAL_BUCKET] { + // A broad query must also find ordinary day-indexed records, including + // records written by older versions. Scan existing entries, not days. + for entry in self.db.iterator_cf(cf, IteratorMode::Start) { let (key, _) = match entry { Ok(pair) => pair, Err(_) => { @@ -1619,9 +1625,6 @@ impl RecordStore for PersistentStore { break; } }; - if !key.starts_with(&prefix) { - break; - } if let Some(record_id) = decode_temporal_record_id(&key) { candidates.insert(record_id); } else { @@ -1629,6 +1632,34 @@ impl RecordStore for PersistentStore { break; } } + } else { + for bucket in buckets + .into_iter() + .chain(std::iter::once(WIDE_TEMPORAL_BUCKET)) + { + let prefix = bucket.to_be_bytes(); + let iter = self + .db + .iterator_cf(cf, IteratorMode::From(&prefix, Direction::Forward)); + for entry in iter { + let (key, _) = match entry { + Ok(pair) => pair, + Err(_) => { + self.mark_read_fault(); + break; + } + }; + if !key.starts_with(&prefix) { + break; + } + if let Some(record_id) = decode_temporal_record_id(&key) { + candidates.insert(record_id); + } else { + self.mark_read_fault(); + break; + } + } + } } let mut records = Vec::new(); for record_id in candidates { @@ -2462,6 +2493,9 @@ fn buckets_for_interval(start: i64, end: i64) -> Vec { let mut buckets = Vec::new(); let mut current = start.div_euclid(TEMPORAL_BUCKET_SECONDS); let end_bucket = (end - 1).div_euclid(TEMPORAL_BUCKET_SECONDS); + if end_bucket - current >= MAX_TEMPORAL_BUCKETS { + return vec![WIDE_TEMPORAL_BUCKET]; + } while current <= end_bucket { buckets.push(current); current += 1; @@ -3417,6 +3451,27 @@ mod tests { .unwrap_or_else(std::sync::PoisonError::into_inner) } + #[test] + fn temporal_bucket_expansion_is_bounded() { + assert_eq!(buckets_for_interval(0, 0), Vec::::new()); + assert_eq!(buckets_for_interval(-1, 1), vec![-1, 0]); + assert_eq!(buckets_for_interval(0, TEMPORAL_BUCKET_SECONDS), vec![0]); + assert_eq!( + buckets_for_interval(0, MAX_TEMPORAL_BUCKETS * TEMPORAL_BUCKET_SECONDS).len(), + MAX_TEMPORAL_BUCKETS as usize + ); + assert_eq!( + buckets_for_interval(0, MAX_TEMPORAL_BUCKETS * TEMPORAL_BUCKET_SECONDS + 1), + vec![WIDE_TEMPORAL_BUCKET] + ); + assert_eq!( + buckets_for_interval(i64::MIN, i64::MAX), + vec![WIDE_TEMPORAL_BUCKET] + ); + assert_eq!(buckets_for_interval(i64::MIN, i64::MIN + 1).len(), 1); + assert_eq!(buckets_for_interval(i64::MAX - 1, i64::MAX).len(), 1); + } + #[test] fn persistent_store_round_trip() { let _guard = lock_persistent_tests(); diff --git a/src/temporal.rs b/src/temporal.rs index d847819..801ac76 100644 --- a/src/temporal.rs +++ b/src/temporal.rs @@ -90,12 +90,13 @@ impl Interval { } /// Get the duration of this interval in seconds - /// Returns None for intervals with infinite endpoints + /// Returns None for intervals with infinite endpoints. Finite durations that + /// exceed i64::MAX seconds saturate at i64::MAX. pub fn duration(&self) -> Option { if self.start == NEG_INF || self.end == POS_INF { None } else { - Some(self.end - self.start) + Some(self.end.saturating_sub(self.start)) } } @@ -107,13 +108,14 @@ impl Interval { } /// Calculate the overlap duration between this interval and another. - /// Returns 0 if the intervals don't overlap. + /// Returns 0 if the intervals don't overlap. Durations exceeding i64::MAX + /// seconds saturate at i64::MAX, including overlap across infinite endpoints. #[inline] pub fn overlap_duration(&self, other: &Interval) -> i64 { let overlap_start = self.start.max(other.start); let overlap_end = self.end.min(other.end); if overlap_start < overlap_end { - overlap_end - overlap_start + overlap_end.saturating_sub(overlap_start) } else { 0 } @@ -164,7 +166,7 @@ impl Ord for Interval { /// All relations are mutually exclusive and collectively exhaustive. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum AllenRelation { - /// a precedes b: a.end <= b.start + /// a precedes b: a.end < b.start Precedes, /// a meets b: a.end == b.start (adjacent) Meets, @@ -182,13 +184,13 @@ pub enum AllenRelation { StartedBy, /// a contains b: a.start < b.start && b.end < a.end Contains, - /// a finished by b: b.start < a.start && b.end == a.end + /// a finished by b: a.start < b.start && a.end == b.end FinishedBy, /// a overlapped by b: b.start < a.start && a.start < b.end < a.end OverlappedBy, /// a met by b: b.end == a.start MetBy, - /// a preceded by b: b.end <= a.start + /// a preceded by b: b.end < a.start PrecededBy, } @@ -200,7 +202,7 @@ pub fn allen_relation(a: &Interval, b: &Interval) -> AllenRelation { (Ordering::Equal, Ordering::Equal) => Equals, (Ordering::Equal, Ordering::Less) => Starts, (Ordering::Equal, Ordering::Greater) => StartedBy, - (Ordering::Less, Ordering::Equal) => Finishes, + (Ordering::Less, Ordering::Equal) => FinishedBy, (Ordering::Less, Ordering::Less) => { if a.end < b.start { Precedes @@ -212,7 +214,7 @@ pub fn allen_relation(a: &Interval, b: &Interval) -> AllenRelation { Contains } } - (Ordering::Greater, Ordering::Equal) => FinishedBy, + (Ordering::Greater, Ordering::Equal) => Finishes, (Ordering::Greater, Ordering::Greater) => { if b.end < a.start { PrecededBy @@ -431,14 +433,65 @@ mod tests { #[test] fn test_allen_relations() { + use AllenRelation::*; + let a = Interval::new(100, 200).unwrap(); - let b = Interval::new(150, 250).unwrap(); - let c = Interval::new(200, 300).unwrap(); - let d = Interval::new(50, 100).unwrap(); + for (start, end, expected, inverse) in [ + (250, 300, Precedes, PrecededBy), + (200, 300, Meets, MetBy), + (150, 250, Overlaps, OverlappedBy), + (100, 250, Starts, StartedBy), + (50, 250, During, Contains), + (50, 200, Finishes, FinishedBy), + (100, 200, Equals, Equals), + ] { + let b = Interval::new(start, end).unwrap(); + assert_eq!(allen_relation(&a, &b), expected); + assert_eq!(allen_relation(&b, &a), inverse); + } + } - assert_eq!(allen_relation(&a, &b), AllenRelation::Overlaps); - assert_eq!(allen_relation(&a, &c), AllenRelation::Meets); - assert_eq!(allen_relation(&a, &d), AllenRelation::MetBy); // d meets a, so a is met by d + #[test] + fn test_overlap_duration_extremes() { + let all = Interval::all_time(); + for (a, b, expected) in [ + (all, all, i64::MAX), + (all, Interval::from_start(-100), i64::MAX), + (all, Interval::until_end(100), i64::MAX), + (Interval::from_start(100), Interval::until_end(200), 100), + (all, Interval::new(-100, 100).unwrap(), 200), + ( + Interval::new(0, 100).unwrap(), + Interval::new(50, 150).unwrap(), + 50, + ), + (Interval::until_end(100), Interval::from_start(100), 0), + (Interval::until_end(100), Interval::from_start(200), 0), + ( + Interval::new(i64::MIN + 1, i64::MAX - 1).unwrap(), + all, + i64::MAX, + ), + ] { + assert_eq!(a.overlap_duration(&b), expected, "{a} and {b}"); + assert_eq!(b.overlap_duration(&a), expected, "{b} and {a}"); + } + } + + #[test] + fn test_duration_extremes() { + for interval in [ + Interval::all_time(), + Interval::from_start(-100), + Interval::until_end(100), + ] { + assert_eq!(interval.duration(), None); + assert_eq!(interval.duration_or_zero(), 0); + } + let wide = Interval::new(i64::MIN + 1, i64::MAX - 1).unwrap(); + assert_eq!(wide.duration(), Some(i64::MAX)); + assert_eq!(wide.duration_or_zero(), i64::MAX); + assert_eq!(Interval::new(-100, 100).unwrap().duration(), Some(200)); } #[test] diff --git a/tests/distributed_e2e.rs b/tests/distributed_e2e.rs index d84883c..ed0af5b 100644 --- a/tests/distributed_e2e.rs +++ b/tests/distributed_e2e.rs @@ -158,6 +158,62 @@ fn normalize_remote_query( } } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn distributed_ingest_and_query_unbounded_intervals() -> anyhow::Result<()> { + let config = support::build_iam_config(); + let (shard0, shard0_handle) = spawn_shard(0, config.clone()).await?; + let (shard1, shard1_handle) = spawn_shard(1, config.clone()).await?; + let (router, router_handle) = spawn_router(vec![shard0, shard1], config).await?; + let mut client = RouterServiceClient::connect(format!("http://{router}")).await?; + + // Exercise partitioned batch ingestion, then match a later batch against its + // persisted unbounded descriptors through the router. + for batch in 0..2 { + let records = (0..100) + .map(|index| { + record_input( + index, + "person", + "crm", + &format!("unbounded-{batch}-{index}"), + vec![("email", "unbounded@example.com", i64::MIN, i64::MAX)], + ) + }) + .collect(); + let response = client + .ingest_records(IngestRecordsRequest { + internal_protocol_version: 5, + records, + }) + .await? + .into_inner(); + assert_eq!(response.assignments.len(), 100); + } + + for (start, end) in [(0, 100), (i64::MIN, i64::MAX)] { + let response = client + .query_entities(proto::QueryEntitiesRequest { + descriptors: vec![proto::QueryDescriptor { + attr: "email".into(), + value: "unbounded@example.com".into(), + }], + start, + end, + }) + .await? + .into_inner(); + assert_eq!( + normalize_remote_query(response), + ("matches", 1, (start, end)) + ); + } + + router_handle.abort(); + shard0_handle.abort(); + shard1_handle.abort(); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn distributed_stream_and_query() -> anyhow::Result<()> { let config = support::build_iam_config(); diff --git a/tests/temporal_sampling.rs b/tests/temporal_sampling.rs new file mode 100644 index 0000000..fefe248 --- /dev/null +++ b/tests/temporal_sampling.rs @@ -0,0 +1,117 @@ +use tempfile::tempdir; +use unirust_rs::ontology::IdentityKey; +use unirust_rs::store::RecordStore; +use unirust_rs::{ + Descriptor, Interval, Ontology, PersistentStore, Record, RecordId, RecordIdentity, + StreamingTuning, Unirust, +}; + +#[test] +fn stochastic_sampling_links_unbounded_and_wide_intervals() -> anyhow::Result<()> { + for interval in [ + Interval::all_time(), + Interval::from_start(-100), + Interval::until_end(100), + Interval::new(i64::MIN + 1, i64::MAX - 1)?, + Interval::new(0, 100)?, + ] { + let dir = tempdir()?; + let store = PersistentStore::open(dir.path())?; + let mut ontology = Ontology::new(); + ontology.add_identity_key(IdentityKey::from_names(vec!["name"], "name")); + // Force sampling even for a single candidate, with full-overlap acceptance + // probability 1, so this regression does not depend on chance or block size. + let tuning = StreamingTuning { + stochastic_sampling: true, + sampling_threshold: 0, + sampling_target: 1, + deferred_reconciliation: false, + ..StreamingTuning::balanced() + }; + let mut engine = Unirust::with_store_and_tuning(ontology, store, tuning); + let attr = engine.intern_attr("name"); + let value = engine.intern_value("Smith"); + let mut assignments = Vec::new(); + for uid in ["first", "second"] { + let record = Record::new( + RecordId(0), + RecordIdentity::new("person".into(), "crm".into(), uid.into()), + vec![Descriptor::new(attr, value, interval)], + ); + assignments.extend(engine.stream_records(vec![record])?); + } + assert_eq!(engine.linker_metrics_snapshot().stochastic_samples, 1); + assert_eq!( + assignments[0].cluster_id, assignments[1].cluster_id, + "{interval}" + ); + assert_eq!(engine.streaming_cluster_count(), Some(1)); + assert_eq!(engine.record_count(), 2); + } + Ok(()) +} + +#[test] +fn persistent_temporal_queries_find_wide_intervals_after_restart() -> anyhow::Result<()> { + let dir = tempdir()?; + { + let store = PersistentStore::open(dir.path())?; + let mut ontology = Ontology::new(); + ontology.add_identity_key(IdentityKey::from_names(vec!["name"], "name")); + let mut engine = Unirust::with_store(ontology, store); + let attr = engine.intern_attr("name"); + for (uid, interval) in [ + ("all", Interval::all_time()), + ("from", Interval::from_start(-100)), + ("until", Interval::until_end(100)), + ("wide", Interval::new(i64::MIN + 1, i64::MAX - 1)?), + ("finite", Interval::new(0, 100)?), + ("future", Interval::new(400_000_000, 800_000_000)?), + ("past", Interval::new(-800_000_000, -400_000_000)?), + ] { + let value = engine.intern_value(uid); + let record = Record::new( + RecordId(0), + RecordIdentity::new("person".into(), "crm".into(), uid.into()), + vec![Descriptor::new(attr, value, interval)], + ); + engine.stream_records(vec![record])?; + } + } + let store = PersistentStore::open(dir.path())?; + for (interval, expected) in [ + ( + Interval::new(0, 100)?, + vec!["all", "finite", "from", "until", "wide"], + ), + (Interval::new(100, 200)?, vec!["all", "from", "wide"]), + (Interval::new(i64::MIN, i64::MIN + 1)?, vec!["all", "until"]), + (Interval::new(i64::MAX - 1, i64::MAX)?, vec!["all", "from"]), + ( + Interval::new(500_000_000, 500_000_001)?, + vec!["all", "from", "future", "wide"], + ), + ( + Interval::new(-500_000_000, -499_999_999)?, + vec!["all", "past", "until", "wide"], + ), + ( + Interval::all_time(), + vec!["all", "finite", "from", "future", "past", "until", "wide"], + ), + ( + Interval::new(-500_000_000, 500_000_000)?, + vec!["all", "finite", "from", "future", "past", "until", "wide"], + ), + (Interval { start: 0, end: 0 }, vec![]), + ] { + let mut actual = store + .get_records_in_interval(interval) + .into_iter() + .map(|record| record.identity.uid) + .collect::>(); + actual.sort(); + assert_eq!(actual, expected, "{interval}"); + } + Ok(()) +}