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 index f47a032524c..e94da008500 100644 --- 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 @@ -35,6 +35,40 @@ impl PostgresStore { &self, history: &EntityHistory, ) -> Result<(), Report> { + let removed: Vec<_> = history.removed().cloned().collect(); + if !removed.is_empty() { + let parameters = EntityTemporalMetadataRow::columnar_parameters(&removed); + self.as_client() + .execute_raw( + " + DELETE FROM entity_temporal_metadata AS temporal + USING + unnest( + $1::UUID [], + $2::UUID [], + $3::UUID [], + $4::UUID [], + $5::TSTZRANGE [], + $6::TSTZRANGE [] + ) + AS previous ( + web_id, entity_uuid, draft_id, edition_id, + decision_time, transaction_time + ) + WHERE temporal.web_id = previous.web_id + AND temporal.entity_uuid = previous.entity_uuid + AND temporal.draft_id IS NOT DISTINCT FROM previous.draft_id + AND temporal.entity_edition_id = previous.edition_id + AND temporal.decision_time = previous.decision_time + AND temporal.transaction_time = previous.transaction_time + ", + parameters + .into_iter() + .map(|(_, column)| column.into_values()), + ) + .await + .change_context(UpdateError)?; + } self.update_entity_temporal_rows( history .updated() 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 index 0460798f32d..cd9cc8b4ca9 100644 --- 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 @@ -15,16 +15,18 @@ struct TemporalEntry { original: Option, } -/// Tracks the final replacement of each loaded row while retaining earlier history +/// Tracks the final replacement or removal of each loaded row while retaining earlier history /// across both time axes. pub(crate) struct EntityHistory { current: HashMap>, + removed: Vec, } impl EntityHistory { pub(super) fn new(rows: impl IntoIterator) -> Self { let mut history = Self { current: HashMap::new(), + removed: Vec::new(), }; for row in rows { history.push(TemporalEntry { @@ -138,6 +140,31 @@ impl EntityHistory { self.push(entry); } + /// Ends the live state that a published draft supersedes while retaining its past on both + /// time axes. A row is deleted only when neither earlier interval survives. + pub(super) fn close( + &mut self, + row: &EntityTemporalMetadataRow, + transaction_time: Timestamp, + decision_time: Timestamp, + ) { + let mut entry = self.take(row); + let mut fragments = Self::split(row, transaction_time, decision_time); + if let Some(row) = fragments.pop() { + entry.row = row; + self.push(entry); + } else { + self.removed.extend(entry.original); + } + for fragment in fragments { + self.insert(fragment); + } + } + + pub(super) fn removed(&self) -> impl Iterator { + self.removed.iter() + } + /// 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( @@ -287,4 +314,25 @@ mod tests { ); } } + + #[test] + fn close_same_transaction_start() { + let original = row(); + let mut history = EntityHistory::new([original.clone()]); + history.close( + &original, + Timestamp::from_unix_timestamp(0), + Timestamp::from_unix_timestamp(0), + ); + assert_eq!( + history.inserted().count(), + 0, + "should omit empty historical intervals" + ); + assert_eq!( + history.removed().count(), + 1, + "should remove the superseded row" + ); + } } 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 6c75b699b44..8cca3ec171a 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 @@ -2898,8 +2898,8 @@ where Ok(temporal_versioning) } + /// Closes the superseded live interval and records the actor responsible for publication. #[tracing::instrument(level = "info", skip(self))] - #[expect(clippy::too_many_lines)] async fn archive_entity( &self, actor_id: ActorId, @@ -2907,148 +2907,24 @@ where transaction_time: Timestamp, decision_time: Timestamp, ) -> Result<(), Report> { - // Closing at the start would leave an empty decision interval - if locked_row.decision_time.start() == &ClosedTemporalBound::Inclusive(decision_time) { - self.as_client() - .query( - "DELETE FROM entity_temporal_metadata - WHERE web_id = $1 - AND entity_uuid = $2 - AND draft_id IS NOT DISTINCT FROM $3 - AND transaction_time @> $4::timestamptz - AND decision_time @> $5::timestamptz;", - &[ - &locked_row.entity_id.web_id, - &locked_row.entity_id.entity_uuid, - &locked_row.entity_id.draft_id, - &transaction_time, - &decision_time, - ], - ) - .instrument(tracing::info_span!( - "DELETE", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres", - )) - .await - .change_context(UpdateError)?; - } else if let Some(draft_id) = locked_row.entity_id.draft_id { - self.client - .as_client() - .query( - " - UPDATE entity_temporal_metadata - SET transaction_time = tstzrange($4::timestamptz, NULL, '[)'), - decision_time = tstzrange(lower(decision_time), $5::timestamptz, '[)') - 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;", - &[ - &locked_row.entity_id.web_id, - &locked_row.entity_id.entity_uuid, - &draft_id, - &transaction_time, - &decision_time, - ], - ) - .instrument(tracing::info_span!( - "UPDATE", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres" - )) - .await - .change_context(UpdateError)?; - } else { - self.client - .as_client() - .query( - " - UPDATE entity_temporal_metadata - SET transaction_time = tstzrange($3::timestamptz, NULL, '[)'), - decision_time = tstzrange(lower(decision_time), $4::timestamptz, '[)') - 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;", - &[ - &locked_row.entity_id.web_id, - &locked_row.entity_id.entity_uuid, - &transaction_time, - &decision_time, - ], - ) - .instrument(tracing::info_span!( - "UPDATE", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres" - )) - .await - .change_context(UpdateError)?; - } - - self.client - .as_client() - .query( + let previous = EntityTemporalMetadataRow::from(locked_row); + let mut history = EntityHistory::new([previous.clone()]); + history.close(&previous, transaction_time, decision_time); + self.update_entity_history(&history).await?; + self.as_client() + .execute( " - INSERT INTO entity_temporal_metadata ( - web_id, - entity_uuid, - draft_id, - entity_edition_id, - transaction_time, - decision_time - ) SELECT - $1, - $2, - $3, - $4, - tstzrange(lower($6::tstzrange), $7, '[)'), - $5 - WHERE lower($6::tstzrange) < $7::timestamptz;", + UPDATE entity_editions + SET provenance = provenance || jsonb_build_object('archivedById', $2::UUID) + WHERE entity_edition_id = $1 + ", &[ - &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, + &previous.entity_edition_id, + &ActorEntityUuid::from(actor_id), ], ) - .instrument(tracing::info_span!( - "INSERT", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres", - )) .await .change_context(UpdateError)?; - - self.as_client() - .query( - " - UPDATE entity_editions SET - provenance = provenance || JSONB_BUILD_OBJECT( - 'archivedById', $2::UUID - ) - WHERE entity_edition_id = $1", - &[&locked_row.entity_edition_id, &actor_id], - ) - .instrument(tracing::info_span!( - "UPDATE", - otel.kind = "client", - db.system = "postgresql", - peer.service = "Postgres", - )) - .await - .change_context(UpdateError)?; - Ok(()) } }