Skip to content

Commit 14f9f26

Browse files
JannisJannis Pohlmann
Jannis
authored and
Jannis Pohlmann
committed
store/postgres: Add overwrite flag to update
This allows to either merge new entity data in at the database level, which is needed for `apply_update_operation`, or to completely overwrite the previous entity data, which is needed for the current implementation of `apply_set_operation` (because that already merges the entity data in and we don't want to do it twice; plus, it behaves differently when it comes to `null` values in entity fields).
1 parent 50c27f8 commit 14f9f26

File tree

3 files changed

+91
-46
lines changed

3 files changed

+91
-46
lines changed

‎store/postgres/src/entities.rs

Lines changed: 72 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -316,11 +316,12 @@ impl<'a> Connection<'a> {
316316
&self,
317317
key: &EntityKey,
318318
data: &serde_json::Value,
319+
overwrite: bool,
319320
guard: Option<EntityFilter>,
320321
history_event: Option<&HistoryEvent>,
321322
) -> Result<usize, StoreError> {
322323
let table = self.table(&key.subgraph_id)?;
323-
table.update(self.conn, key, data, guard, history_event)
324+
table.update(self.conn, key, data, overwrite, guard, history_event)
324325
}
325326

326327
pub(crate) fn delete(
@@ -400,6 +401,7 @@ impl SplitTable {
400401
conn: &PgConnection,
401402
key: &EntityKey,
402403
data: &serde_json::Value,
404+
overwrite: bool,
403405
guard: Option<EntityFilter>,
404406
history_event: Option<&HistoryEvent>,
405407
) -> Result<usize, StoreError> {
@@ -430,23 +432,42 @@ impl SplitTable {
430432
.filter(entities::entity.eq(&key.entity_type))
431433
.filter(entities::id.eq(&key.entity_id));
432434

433-
Ok(diesel::update(target)
434-
.set((
435-
entities::data.eq(entities::data.merge(&data)),
436-
entities::event_source.eq(&event_source),
437-
))
438-
.filter(filter)
439-
.execute(conn)?)
435+
if overwrite {
436+
Ok(diesel::update(target)
437+
.set((
438+
entities::data.eq(&data),
439+
entities::event_source.eq(&event_source),
440+
))
441+
.filter(filter)
442+
.execute(conn)?)
443+
} else {
444+
Ok(diesel::update(target)
445+
.set((
446+
entities::data.eq(entities::data.merge(&data)),
447+
entities::event_source.eq(&event_source),
448+
))
449+
.filter(filter)
450+
.execute(conn)?)
451+
}
440452
} else {
441453
// If there is no guard (which has to include all 'normal' subgraphs),
442454
// we need to use a direct query since diesel::update does not like
443455
// dynamic tables.
444-
let query = format!(
445-
"update {}.entities
446-
set data = data || $3, event_source = $4
447-
where entity = $1 and id = $2",
448-
self.schema
449-
);
456+
let query = if overwrite {
457+
format!(
458+
"update {}.entities
459+
set data = $3, event_source = $4
460+
where entity = $1 and id = $2",
461+
self.schema
462+
)
463+
} else {
464+
format!(
465+
"update {}.entities
466+
set data = data || $3, event_source = $4
467+
where entity = $1 and id = $2",
468+
self.schema
469+
)
470+
};
450471
let query = diesel::sql_query(query)
451472
.bind::<Text, _>(&key.entity_type)
452473
.bind::<Text, _>(&key.entity_id)
@@ -746,6 +767,7 @@ impl Table {
746767
conn: &PgConnection,
747768
key: &EntityKey,
748769
data: &serde_json::Value,
770+
overwrite: bool,
749771
guard: Option<EntityFilter>,
750772
history_event: Option<&HistoryEvent>,
751773
) -> Result<usize, StoreError> {
@@ -758,22 +780,42 @@ impl Table {
758780
.filter(public::entities::entity.eq(&key.entity_type))
759781
.filter(public::entities::id.eq(&key.entity_id));
760782

761-
let query = diesel::update(target).set((
762-
public::entities::data.eq(public::entities::data.merge(data)),
763-
public::entities::event_source.eq(&event_source),
764-
));
765-
766-
match guard {
767-
Some(filter) => {
768-
let filter = build_filter(filter).map_err(|e| {
769-
TransactionAbortError::Other(format!(
770-
"invalid filter '{}' for value '{}'",
771-
e.filter, e.value
772-
))
773-
})?;
774-
Ok(query.filter(filter).execute(conn)?)
783+
if overwrite {
784+
let query = diesel::update(target).set((
785+
public::entities::data.eq(data),
786+
public::entities::event_source.eq(&event_source),
787+
));
788+
789+
match guard {
790+
Some(filter) => {
791+
let filter = build_filter(filter).map_err(|e| {
792+
TransactionAbortError::Other(format!(
793+
"invalid filter '{}' for value '{}'",
794+
e.filter, e.value
795+
))
796+
})?;
797+
Ok(query.filter(filter).execute(conn)?)
798+
}
799+
None => Ok(query.execute(conn)?),
800+
}
801+
} else {
802+
let query = diesel::update(target).set((
803+
public::entities::data.eq(public::entities::data.merge(data)),
804+
public::entities::event_source.eq(&event_source),
805+
));
806+
807+
match guard {
808+
Some(filter) => {
809+
let filter = build_filter(filter).map_err(|e| {
810+
TransactionAbortError::Other(format!(
811+
"invalid filter '{}' for value '{}'",
812+
e.filter, e.value
813+
))
814+
})?;
815+
Ok(query.filter(filter).execute(conn)?)
816+
}
817+
None => Ok(query.execute(conn)?),
775818
}
776-
None => Ok(query.execute(conn)?),
777819
}
778820
}
779821
Table::Split(entities) => {
@@ -791,7 +833,7 @@ impl Table {
791833
)?;
792834
}
793835

794-
entities.update(conn, key, data, guard, history_event)
836+
entities.update(conn, key, data, overwrite, guard, history_event)
795837
}
796838
}
797839
}

‎store/postgres/src/store.rs

Lines changed: 17 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -442,28 +442,31 @@ impl Store {
442442
Some(_) => (true, 0),
443443
};
444444

445-
// Apply the operation
445+
// Create set operation
446446
let operation = EntityOperation::Set {
447447
key: key.clone(),
448448
data,
449449
};
450+
451+
// Apply the operation to obtain the updated (or new) entity
450452
let updated_entity = operation.apply(existing_entity)?;
451-
let updated_json: serde_json::Value =
452-
serde_json::to_value(&updated_entity).map_err(|e| {
453-
format_err!(
454-
"Failed to set entity ({}, {}, {}) as setting it would break it: {}",
455-
key.subgraph_id,
456-
key.entity_type,
457-
key.entity_id,
458-
e
459-
)
460-
})?;
453+
454+
// Convert the entity data to JSON
455+
let json_data: serde_json::Value = serde_json::to_value(updated_entity).map_err(|e| {
456+
format_err!(
457+
"Failed to set entity ({}, {}, {}) as setting it would break it: {}",
458+
key.subgraph_id,
459+
key.entity_type,
460+
key.entity_id,
461+
e
462+
)
463+
})?;
461464

462465
// Either insert or update the entity in Postgres
463466
let result = if is_update {
464-
conn.update(&key, &updated_json, None, history_event)
467+
conn.update(&key, &json_data, true, None, history_event)
465468
} else {
466-
conn.insert(&key, &updated_json, history_event)
469+
conn.insert(&key, &json_data, history_event)
467470
};
468471

469472
result.map(|_| count).map_err(|e| {
@@ -500,7 +503,7 @@ impl Store {
500503
})?;
501504

502505
// Update the entity in Postgres
503-
match conn.update(&key, &json, guard, history_event)? {
506+
match conn.update(&key, &json, false, guard, history_event)? {
504507
0 => Err(TransactionAbortError::AbortUnless {
505508
expected_entity_ids: vec![key.entity_id.clone()],
506509
actual_entity_ids: vec![],

‎store/postgres/tests/store.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -318,7 +318,7 @@ fn get_entity_1() {
318318
);
319319
expected_entity.insert("weight".to_owned(), Value::BigDecimal(184.4.into()));
320320
expected_entity.insert("coffee".to_owned(), Value::Bool(false));
321-
// favorite_color was null, so we expect the property to be omitted
321+
// "favorite_color" was set to `Null` earlier and should be absent
322322

323323
// Check that the expected entity was returned
324324
assert_eq!(result, Some(expected_entity));
@@ -355,7 +355,7 @@ fn get_entity_3() {
355355
);
356356
expected_entity.insert("weight".to_owned(), Value::BigDecimal(111.7.into()));
357357
expected_entity.insert("coffee".to_owned(), Value::Bool(false));
358-
// favorite_color was later set to null, so we expect the property to be omitted
358+
// "favorite_color" was set to `Null` earlier and should be absent
359359

360360
// Check that the expected entity was returned
361361
assert_eq!(result, Some(expected_entity));

0 commit comments

Comments
 (0)