store: ON CONFLICT DO NOTHING for skip_duplicates inserts

Add conditional ON CONFLICT (primary_key) DO NOTHING clause to
InsertQuery::walk_ast() when table.immutable && table.skip_duplicates.
This handles cross-batch duplicates where an entity committed in a
previous batch is re-inserted due to the immutable entity cache
skipping store lookups.

Two unit tests verify: skip_duplicates inserts include ON CONFLICT,
default immutable inserts do not.

Author: isum

store/postgres/src/relational/query_tests.rs

@@ -2,16 +2,19 @@ use std::{collections::BTreeSet, sync::Arc};
 
 use diesel::{debug_query, pg::Pg};
 use graph::{
+    components::store::write::RowGroup,
     data_source::CausalityRegion,
-    prelude::{r, serde_json as json, DeploymentHash, EntityFilter},
+    entity,
+    prelude::{r, serde_json as json, DeploymentHash, EntityFilter, Logger},
     schema::InputSchema,
+    slog,
 };
 
 use crate::{
     block_range::BoundSide,
     layout_for_tests::{make_dummy_site, Namespace},
     relational::{Catalog, ColumnType, Layout},
-    relational_queries::{FindRangeQuery, FromColumnValue},
+    relational_queries::{FindRangeQuery, FromColumnValue, InsertQuery},
 };
 
 use crate::relational_queries::Filter;
@@ -197,3 +200,64 @@ fn test_id_type_casting(table: &crate::relational::Table, expected_cast: &str, t
         sql
     );
 }
+
+fn insert_sql_for_schema(gql: &str, entity_type_name: &str) -> String {
+    use graph::components::store::write::EntityModification;
+
+    let layout = test_layout(gql);
+    let schema = &layout.input_schema;
+    let et = schema.entity_type(entity_type_name).unwrap();
+    let table = layout.table_for_entity(&et).unwrap();
+
+    let mut entity = entity! { schema => id: "test1" };
+    entity.set_vid(1).unwrap();
+    let key = et.key(graph::data::store::Id::String("test1".into()));
+    let emod = EntityModification::Insert {
+        key,
+        data: Arc::new(entity),
+        block: 1,
+        end: None,
+    };
+
+    let logger = Logger::root(slog::Discard, slog::o!());
+    let mut group = RowGroup::new(et, table.immutable, table.skip_duplicates, logger);
+    group.push(emod, 1).unwrap();
+
+    let chunks: Vec<_> = group.write_chunks(100).collect();
+    let chunk = &chunks[0];
+    let query = InsertQuery::new(table.as_ref(), chunk).unwrap();
+    debug_query::<Pg, _>(&query).to_string()
+}
+
+#[test]
+fn skip_duplicates_insert_generates_on_conflict() {
+    let schema = "
+    type Thing @entity(immutable: true, skipDuplicates: true) {
+        id: String!
+    }";
+    let sql = insert_sql_for_schema(schema, "Thing");
+    assert!(
+        sql.contains("ON CONFLICT"),
+        "Expected ON CONFLICT in SQL for skip_duplicates immutable table, got: {}",
+        sql
+    );
+    assert!(
+        sql.contains("DO NOTHING"),
+        "Expected DO NOTHING in SQL for skip_duplicates immutable table, got: {}",
+        sql
+    );
+}
+
+#[test]
+fn default_immutable_insert_has_no_on_conflict_skip_duplicates() {
+    let schema = "
+    type Thing @entity(immutable: true) {
+        id: String!
+    }";
+    let sql = insert_sql_for_schema(schema, "Thing");
+    assert!(
+        !sql.contains("ON CONFLICT"),
+        "Default immutable table should NOT have ON CONFLICT, got: {}",
+        sql
+    );
+}

store/postgres/src/relational_queries.rs

@@ -2584,6 +2584,12 @@ impl<'a> QueryFragment<Pg> for InsertQuery<'a> {
             out.push_sql(")");
         }
 
+        if self.table.immutable && self.table.skip_duplicates {
+            out.push_sql("\n ON CONFLICT (");
+            out.push_identifier(self.table.primary_key().name.as_str())?;
+            out.push_sql(") DO NOTHING");
+        }
+
         Ok(())
     }
 }