diff --git a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/batch/mod.rs b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/batch/mod.rs new file mode 100644 index 00000000000..f2215a879c5 --- /dev/null +++ b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/batch/mod.rs @@ -0,0 +1 @@ +mod write; diff --git a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/batch/write.rs b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/batch/write.rs new file mode 100644 index 00000000000..f47a032524c --- /dev/null +++ b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/batch/write.rs @@ -0,0 +1,107 @@ +use error_stack::{Report, ResultExt as _}; +use hash_graph_store::error::UpdateError; +use tokio_postgres::GenericClient as _; + +use crate::store::{ + AsClient, InTransaction, PostgresStore, + postgres::{ + knowledge::entity::history::EntityHistory, + query::{ + bulk_insert, + rows::{EntityTemporalMetadataRow, PostgresRow}, + }, + }, +}; + +impl PostgresStore { + async fn insert_rows( + &self, + rows: &[R], + ) -> Result<(), Report> { + if rows.is_empty() { + return Ok(()); + } + let (statement, parameters) = bulk_insert().rows(rows).compile(); + self.as_client() + .execute_raw(&statement, parameters) + .await + .change_context(UpdateError)?; + Ok(()) + } + + /// Applies the final temporal diff while preserving the non-overlap constraints. + /// Existing live intervals must be closed before draft rows can become live. + pub(crate) async fn update_entity_history( + &self, + history: &EntityHistory, + ) -> Result<(), Report> { + self.update_entity_temporal_rows( + history + .updated() + .filter(|(original, row)| original.draft_id == row.draft_id), + ) + .await?; + self.update_entity_temporal_rows( + history + .updated() + .filter(|(original, row)| original.draft_id != row.draft_id), + ) + .await?; + let inserted: Vec<_> = history.inserted().cloned().collect(); + self.insert_rows(&inserted).await + } + + /// Matches updates against the full original interval because an edition can occupy multiple + /// temporal rows. Each original row must occur only once in the input. + async fn update_entity_temporal_rows<'a>( + &self, + rows: impl Iterator + + Send, + ) -> Result<(), Report> { + let mut previous = Vec::new(); + let mut updated = Vec::new(); + for (original, row) in rows { + previous.push(original.clone()); + updated.push(row.clone()); + } + if !updated.is_empty() { + let mut parameters = EntityTemporalMetadataRow::columnar_parameters(&previous); + parameters.extend(EntityTemporalMetadataRow::columnar_parameters(&updated)); + self.as_client() + .execute_raw( + " + UPDATE entity_temporal_metadata AS temporal + SET + draft_id = updates.draft_id, + entity_edition_id = updates.edition_id, + decision_time = updates.decision_time, + transaction_time = updates.transaction_time + FROM + unnest( + $1::UUID [], $2::UUID [], $3::UUID [], $4::UUID [], + $5::TSTZRANGE [], $6::TSTZRANGE [], + $7::UUID [], $8::UUID [], $9::UUID [], $10::UUID [], + $11::TSTZRANGE [], $12::TSTZRANGE [] + ) AS updates ( + previous_web_id, previous_entity_uuid, previous_draft_id, + previous_edition_id, previous_decision_time, previous_transaction_time, + web_id, entity_uuid, draft_id, edition_id, + decision_time, transaction_time + ) + WHERE temporal.web_id = updates.previous_web_id + AND temporal.entity_uuid = updates.previous_entity_uuid + AND temporal.draft_id IS NOT DISTINCT FROM updates.previous_draft_id + AND temporal.entity_edition_id = updates.previous_edition_id + AND temporal.decision_time = updates.previous_decision_time + AND temporal.transaction_time = updates.previous_transaction_time + ", + parameters + .into_iter() + .map(|(_, column)| column.into_values()), + ) + .await + .change_context(UpdateError)?; + } + Ok(()) + } +} diff --git a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/history.rs b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/history.rs new file mode 100644 index 00000000000..0460798f32d --- /dev/null +++ b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/history.rs @@ -0,0 +1,290 @@ +use std::collections::HashMap; + +use hash_graph_temporal_versioning::{ + ClosedTemporalBound, DecisionTime, LeftClosedTemporalInterval, OpenTemporalBound, Timestamp, + TransactionTime, +}; +use type_system::knowledge::entity::EntityId; + +use crate::store::postgres::query::rows::EntityTemporalMetadataRow; + +/// Carries a row's identity from the start of the batch through repeated edits so persistence can +/// update the original database row once. +struct TemporalEntry { + row: EntityTemporalMetadataRow, + original: Option, +} + +/// Tracks the final replacement of each loaded row while retaining earlier history +/// across both time axes. +pub(crate) struct EntityHistory { + current: HashMap>, +} + +impl EntityHistory { + pub(super) fn new(rows: impl IntoIterator) -> Self { + let mut history = Self { + current: HashMap::new(), + }; + for row in rows { + history.push(TemporalEntry { + original: Some(row.clone()), + row, + }); + } + history + } + + #[cfg(test)] + pub(super) fn get( + &self, + entity_id: EntityId, + transaction_time: Timestamp, + decision_time: Timestamp, + ) -> Option<&EntityTemporalMetadataRow> { + self.current + .get(&entity_id)? + .iter() + .map(|entry| &entry.row) + .find(|row| { + row.transaction_time.contains_point(&transaction_time) + && row.decision_time.contains_point(&decision_time) + }) + } + + fn push(&mut self, entry: TemporalEntry) { + self.current + .entry(EntityId { + web_id: entry.row.web_id, + entity_uuid: entry.row.entity_uuid, + draft_id: entry.row.draft_id, + }) + .or_default() + .push(entry); + } + + fn take(&mut self, row: &EntityTemporalMetadataRow) -> TemporalEntry { + let entries = self + .current + .get_mut(&EntityId { + web_id: row.web_id, + entity_uuid: row.entity_uuid, + draft_id: row.draft_id, + }) + .expect("changed entity should have temporal history"); + let index = entries + .iter() + .position(|entry| entry.row == *row) + .expect("changed temporal row should be present"); + entries.swap_remove(index) + } + + pub(super) fn insert(&mut self, row: EntityTemporalMetadataRow) { + self.push(TemporalEntry { + row, + original: None, + }); + } + + /// Produces intervals that preserve earlier transaction snapshots and earlier decision-time + /// states when a patch replaces the remainder of a row. + fn split( + row: &EntityTemporalMetadataRow, + transaction_time: Timestamp, + decision_time: Timestamp, + ) -> Vec { + let mut fragments = Vec::with_capacity(2); + let ClosedTemporalBound::Inclusive(transaction_start) = *row.transaction_time.start(); + if transaction_start < transaction_time { + fragments.push(EntityTemporalMetadataRow { + transaction_time: LeftClosedTemporalInterval::new( + ClosedTemporalBound::Inclusive(transaction_start), + OpenTemporalBound::Exclusive(transaction_time), + ), + ..row.clone() + }); + } + let ClosedTemporalBound::Inclusive(decision_start) = *row.decision_time.start(); + if decision_start < decision_time { + fragments.push(EntityTemporalMetadataRow { + transaction_time: LeftClosedTemporalInterval::new( + ClosedTemporalBound::Inclusive(transaction_time), + OpenTemporalBound::Unbounded, + ), + decision_time: LeftClosedTemporalInterval::new( + ClosedTemporalBound::Inclusive(decision_start), + OpenTemporalBound::Exclusive(decision_time), + ), + ..row.clone() + }); + } + fragments + } + + /// Keeps the history before the patch on both time axes, then associates the edited interval + /// with the same original row so repeated patches produce one final update. + pub(super) fn replace( + &mut self, + previous: &EntityTemporalMetadataRow, + row: EntityTemporalMetadataRow, + transaction_time: Timestamp, + decision_time: Timestamp, + ) { + let mut entry = self.take(previous); + for fragment in Self::split(previous, transaction_time, decision_time) { + self.insert(fragment); + } + entry.row = row; + self.push(entry); + } + + /// Pairs each changed persisted row with its final replacement. Repeated edits are combined + /// so a bulk update never targets the same database row twice. + pub(super) fn updated( + &self, + ) -> impl Iterator { + self.current.values().flatten().filter_map(|entry| { + let original = entry.original.as_ref()?; + if *original == entry.row { + return None; + } + Some((original, &entry.row)) + }) + } + + /// Includes fragments kept for historical reads as well as intervals for new editions. + pub(super) fn inserted(&self) -> impl Iterator { + self.current + .values() + .flatten() + .filter(|entry| entry.original.is_none()) + .map(|entry| &entry.row) + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + + use hash_graph_temporal_versioning::{ + ClosedTemporalBound, LeftClosedTemporalInterval, OpenTemporalBound, Timestamp, + }; + use type_system::{ + knowledge::entity::{ + EntityId, + id::{EntityEditionId, EntityUuid}, + }, + principal::actor_group::WebId, + }; + use uuid::Uuid; + + use super::EntityHistory; + use crate::store::postgres::query::rows::EntityTemporalMetadataRow; + + fn interval(start: i64, end: Option) -> LeftClosedTemporalInterval { + LeftClosedTemporalInterval::new( + ClosedTemporalBound::Inclusive(Timestamp::from_unix_timestamp(start)), + end.map_or(OpenTemporalBound::Unbounded, |end| { + OpenTemporalBound::Exclusive(Timestamp::from_unix_timestamp(end)) + }), + ) + } + + fn row() -> EntityTemporalMetadataRow { + EntityTemporalMetadataRow { + web_id: WebId::new(Uuid::nil()), + entity_uuid: EntityUuid::new(Uuid::nil()), + draft_id: None, + entity_edition_id: EntityEditionId::new(Uuid::nil()), + decision_time: interval(0, None), + transaction_time: interval(0, None), + } + } + + /// Checks history preservation and a single planned update of the original row. + /// + /// `decisions` contains replacement decision-time starts in application order. Each `expected` + /// tuple is `(decision_start, decision_end, edition)` for rows visible from transaction time 10 + /// onward. Starts are inclusive, ends exclusive, and `None` means unbounded. Edition 0 is the + /// original state; subsequent editions follow replacement order. The transaction snapshot + /// `[0, 10)` is checked separately. + #[test] + fn repeated_row_decision_times() { + for (decisions, expected) in [ + ([5, 5], vec![(0, Some(5), 0), (5, None, 2)]), + ([3, 5], vec![(0, Some(3), 0), (3, Some(5), 1), (5, None, 2)]), + ([5, 3], vec![(0, Some(3), 0), (3, Some(5), 2), (5, None, 1)]), + ([0, 0], vec![(0, None, 2)]), + ] { + let original = row(); + let id = EntityId { + web_id: original.web_id, + entity_uuid: original.entity_uuid, + draft_id: None, + }; + let mut history = EntityHistory::new([original.clone()]); + let transaction_time = Timestamp::from_unix_timestamp(10); + for (index, decision) in decisions.into_iter().enumerate() { + let decision_time = Timestamp::from_unix_timestamp(decision); + let previous = history + .get(id, transaction_time, decision_time) + .expect("target row should exist") + .clone(); + history.replace( + &previous, + EntityTemporalMetadataRow { + entity_edition_id: EntityEditionId::new(Uuid::from_u128(index as u128 + 1)), + decision_time: LeftClosedTemporalInterval::new( + ClosedTemporalBound::Inclusive(decision_time), + *previous.decision_time.end(), + ), + transaction_time: interval(10, None), + ..previous.clone() + }, + transaction_time, + decision_time, + ); + } + let mut expected_rows: HashSet<_> = expected + .into_iter() + .map(|(start, end, edition)| EntityTemporalMetadataRow { + entity_edition_id: EntityEditionId::new(Uuid::from_u128(edition)), + decision_time: interval(start, end), + transaction_time: interval(10, None), + ..original.clone() + }) + .collect(); + expected_rows.insert(EntityTemporalMetadataRow { + transaction_time: interval(0, Some(10)), + ..original + }); + assert_eq!( + history + .current + .values() + .flatten() + .map(|entry| entry.row.clone()) + .collect::>(), + expected_rows, + "should preserve the resolved history for {decisions:?}" + ); + assert_eq!( + history + .updated() + .map(|(original, _)| original) + .collect::>(), + vec![&original], + "should update the original row once" + ); + assert_eq!( + history + .inserted() + .chain(history.updated().map(|(_, row)| row)) + .cloned() + .collect::>(), + expected_rows, + "should persist only resolved rows" + ); + } + } +} diff --git a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/mod.rs b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/mod.rs index 231abc0610d..6c75b699b44 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/mod.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/knowledge/entity/mod.rs @@ -1,5 +1,7 @@ +mod batch; mod delete; pub(crate) mod feed; +mod history; mod patch; pub(crate) mod provenance; mod query; @@ -103,6 +105,7 @@ use crate::store::{ BeginReadOnlyTransaction, InTransaction, TransactionState, TraversalContext, crud::{QueryIndices, TypedRow}, knowledge::entity::{ + history::EntityHistory, patch::{EntityPatchPreparation, PreparedEntityPatch}, provenance::{SqlEntityEditionProvenance, SqlEntityProvenance}, read::EntityEdgeTraversalData, @@ -2485,6 +2488,19 @@ struct LockedEntityEdition { transaction_time: LeftClosedTemporalInterval, } +impl From for EntityTemporalMetadataRow { + fn from(edition: LockedEntityEdition) -> Self { + Self { + web_id: edition.entity_id.web_id, + entity_uuid: edition.entity_id.entity_uuid, + draft_id: edition.entity_id.draft_id, + entity_edition_id: edition.entity_edition_id, + decision_time: edition.decision_time, + transaction_time: edition.transaction_time, + } + } +} + /// Builds the statement populating `entity_edition_cache` by aggregating the editions' /// `entity_is_of_type` rows joined to the referenced types. /// @@ -2848,8 +2864,8 @@ where }) } + /// Preserves the earlier intervals before replacing the state at the patch timestamps. #[tracing::instrument(level = "info", skip(self))] - #[expect(clippy::too_many_lines)] async fn update_temporal_metadata( &self, locked_row: LockedEntityEdition, @@ -2858,186 +2874,28 @@ where entity_edition_id: EntityEditionId, undraft: bool, ) -> Result> { - let row = if let Some(draft_id) = locked_row.entity_id.draft_id { - if undraft { - self.client - .as_client() - .query_one( - " - UPDATE entity_temporal_metadata - SET transaction_time = tstzrange($4::timestamptz, NULL, '[)'), - decision_time = tstzrange($5::timestamptz, upper(decision_time), '[)'), - entity_edition_id = $6, - draft_id = NULL - WHERE entity_temporal_metadata.web_id = $1 - AND entity_temporal_metadata.entity_uuid = $2 - AND entity_temporal_metadata.draft_id = $3 - AND entity_temporal_metadata.transaction_time @> $4::timestamptz - AND entity_temporal_metadata.decision_time @> $5::timestamptz - RETURNING decision_time, transaction_time;", - &[ - &locked_row.entity_id.web_id, - &locked_row.entity_id.entity_uuid, - &draft_id, - &transaction_time, - &decision_time, - &entity_edition_id, - ], - ) - .instrument(tracing::info_span!( - "UPDATE", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres" - )) - .await - .change_context(UpdateError)? - } else { - self.client - .as_client() - .query_one( - " - UPDATE entity_temporal_metadata - SET transaction_time = tstzrange($4::timestamptz, NULL, '[)'), - decision_time = tstzrange($5::timestamptz, upper(decision_time), '[)'), - entity_edition_id = $6 - WHERE entity_temporal_metadata.web_id = $1 - AND entity_temporal_metadata.entity_uuid = $2 - AND entity_temporal_metadata.draft_id = $3 - AND entity_temporal_metadata.transaction_time @> $4::timestamptz - AND entity_temporal_metadata.decision_time @> $5::timestamptz - RETURNING decision_time, transaction_time;", - &[ - &locked_row.entity_id.web_id, - &locked_row.entity_id.entity_uuid, - &draft_id, - &transaction_time, - &decision_time, - &entity_edition_id, - ], - ) - .instrument(tracing::info_span!( - "UPDATE", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres" - )) - .await - .change_context(UpdateError)? - } - } else { - self.client - .as_client() - .query_one( - " - UPDATE entity_temporal_metadata - SET transaction_time = tstzrange($3::timestamptz, NULL, '[)'), - decision_time = tstzrange($4::timestamptz, upper(decision_time), '[)'), - entity_edition_id = $5 - WHERE entity_temporal_metadata.web_id = $1 - AND entity_temporal_metadata.entity_uuid = $2 - AND entity_temporal_metadata.draft_id IS NULL - AND entity_temporal_metadata.transaction_time @> $3::timestamptz - AND entity_temporal_metadata.decision_time @> $4::timestamptz - RETURNING decision_time, transaction_time;", - &[ - &locked_row.entity_id.web_id, - &locked_row.entity_id.entity_uuid, - &transaction_time, - &decision_time, - &entity_edition_id, - ], - ) - .instrument(tracing::info_span!( - "UPDATE", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres" - )) - .await - .change_context(UpdateError)? + let previous = EntityTemporalMetadataRow::from(locked_row); + let row = EntityTemporalMetadataRow { + draft_id: if undraft { None } else { previous.draft_id }, + entity_edition_id, + transaction_time: LeftClosedTemporalInterval::new( + ClosedTemporalBound::Inclusive(transaction_time), + OpenTemporalBound::Unbounded, + ), + decision_time: LeftClosedTemporalInterval::new( + ClosedTemporalBound::Inclusive(decision_time), + *previous.decision_time.end(), + ), + ..previous.clone() }; - - self.client - .as_client() - .query( - " - INSERT INTO entity_temporal_metadata ( - web_id, - entity_uuid, - draft_id, - entity_edition_id, - decision_time, - transaction_time - ) SELECT - $1, - $2, - $3, - $4, - $5, - tstzrange(lower($6::tstzrange), $7, '[)') - WHERE lower($6::tstzrange) < $7::timestamptz;", - &[ - &locked_row.entity_id.web_id, - &locked_row.entity_id.entity_uuid, - &locked_row.entity_id.draft_id, - &locked_row.entity_edition_id, - &locked_row.decision_time, - &locked_row.transaction_time, - &transaction_time, - ], - ) - .instrument(tracing::info_span!( - "INSERT", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres", - )) - .await - .change_context(UpdateError)?; - - self.client - .as_client() - .query( - " - INSERT INTO entity_temporal_metadata ( - web_id, - entity_uuid, - draft_id, - entity_edition_id, - transaction_time, - decision_time - ) SELECT - $1, - $2, - $3, - $4, - tstzrange($6, NULL, '[)'), - tstzrange(lower($5::tstzrange), $7, '[)') - WHERE lower($5::tstzrange) < $7::timestamptz;", - &[ - &locked_row.entity_id.web_id, - &locked_row.entity_id.entity_uuid, - &locked_row.entity_id.draft_id, - &locked_row.entity_edition_id, - &locked_row.decision_time, - &transaction_time, - &decision_time, - ], - ) - .instrument(tracing::info_span!( - "INSERT", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres", - )) - .await - .change_context(UpdateError)?; - - Ok(EntityTemporalMetadata { - decision_time: row.get(0), - transaction_time: row.get(1), - }) + let temporal_versioning = EntityTemporalMetadata { + decision_time: row.decision_time, + transaction_time: row.transaction_time, + }; + let mut history = EntityHistory::new([previous.clone()]); + history.replace(&previous, row, transaction_time, decision_time); + self.update_entity_history(&history).await?; + Ok(temporal_versioning) } #[tracing::instrument(level = "info", skip(self))] diff --git a/libs/@local/graph/postgres-store/src/store/postgres/query/rows.rs b/libs/@local/graph/postgres-store/src/store/postgres/query/rows.rs index c056f0a6150..af429110961 100644 --- a/libs/@local/graph/postgres-store/src/store/postgres/query/rows.rs +++ b/libs/@local/graph/postgres-store/src/store/postgres/query/rows.rs @@ -571,7 +571,7 @@ impl PostgresRow for EntityIsOfTypeRow { } } -#[derive(Debug)] +#[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct EntityTemporalMetadataRow { pub web_id: WebId, pub entity_uuid: EntityUuid,