Add skipDuplicates parameter on @entity directive (#6458)

* graph: parse skipDuplicates from @entity directive

Add kw::SKIP_DUPLICATES constant, skip_duplicates bool field to ObjectType,
and parsing logic in ObjectType::new() defaulting to false when absent.

* graph: validate skipDuplicates directive argument

Added SkipDuplicatesRequiresImmutable error variant, bool_arg validation
for skipDuplicates in validate_entity_directives(), and three test functions
covering non-boolean value, mutable entity, and timeseries+skipDuplicates.

* graph: expose skip_duplicates on EntityType and InputSchema APIs

Add TypeInfo::skip_duplicates(), InputSchema::skip_duplicates(), and
EntityType::skip_duplicates() following the is_immutable() three-layer
delegation pattern. Object types return immutable && skip_duplicates;
Interface and Aggregation types return false.

* graph, store: propagate skip_duplicates to RowGroup

Add skip_duplicates: bool field to RowGroup struct alongside immutable,
update RowGroup::new() to accept the parameter, and wire it from
entity_type.skip_duplicates() in RowGroups::group_entry(). All other
call sites (RowGroupForPerfTest, test helpers, example) pass false.

* graph, store: lenient write-batch enforcement for skip_duplicates

Modify RowGroup::append_row() so that when immutable=true and
skip_duplicates=true, cross-block duplicate inserts and
Overwrite/Remove operations log warnings and return Ok(()) instead
of failing. Same-block duplicates remain allowed. Default behavior
(skip_duplicates=false) is preserved exactly.

Added Logger field to RowGroup/RowGroups with CacheWeight impl,
threaded through all construction sites. 5 unit tests covering all
scenarios.

* store: propagate skip_duplicates to Table

Added skip_duplicates: bool field to Table struct in relational.rs,
wired from EntityType::skip_duplicates() in Table::new(), copied in
Table::new_like(), and defaulted to false in make_poi_table().

* store: lenient store-layer enforcement for skip_duplicates

Added logger parameter to Layout::update() and Layout::delete() in
relational.rs. When table.immutable && table.skip_duplicates, these
methods now log a warning and return Ok(0) instead of returning an
error. Default immutable behavior (skip_duplicates=false) is preserved.
Updated all callers including deployment_store.rs and test files.
Added 4 unit tests with SkipDupMink entity type to verify both
skip_duplicates and default immutable behavior.

* 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.

* store: log cross-batch duplicate inserts in Layout::insert()

Restructured Layout::insert() from if-let-Err to match to capture
affected_rows from InsertQuery::execute(). Tracks expected vs actual
row counts across both the main batch path and the row-by-row fallback
path. Logs a warning when affected_rows < expected for skip_duplicates
immutable tables.

* tests: runner test for skipDuplicates immutable entities

Add end-to-end runner test exercising @entity(immutable: true,
skipDuplicates: true) with duplicate entity inserts across blocks.

- tests/runner-tests/skip-duplicates/: subgraph with Ping entity using
  skipDuplicates, block handler that saves same ID every block
- tests/tests/runner_tests.rs: skip_duplicates() test syncs 4 blocks
  and verifies entity is queryable (indexing did not fail)

* graph, store: address review comments

- Remove logs and the logger from the RowGroup
- Simplify schema tests
- Add ObjectMutability enum instead of two booleans

Author: isum

graph/examples/append_row.rs

@@ -104,7 +104,7 @@ pub fn main() -> anyhow::Result<()> {
             };
             mods.push(md);
         }
-        let mut group = RowGroup::new(THING_TYPE.clone(), false);
+        let mut group = RowGroup::new(THING_TYPE.clone());
 
         let start = Instant::now();
         for md in mods {

graph/src/components/store/write.rs

@@ -4,6 +4,8 @@ use std::{
     sync::Arc,
 };
 
+use slog::Logger;
+
 use crate::{
     blockchain::{block_stream::FirehoseCursor, BlockPtr, BlockTime},
     cheap_clone::CheapClone,
@@ -338,24 +340,22 @@ impl LastMod {
 #[derive(Debug, CacheWeight)]
 pub struct RowGroup {
     pub entity_type: EntityType,
+
     /// All changes for this entity type, ordered by block; i.e., if `i < j`
     /// then `rows[i].block() <= rows[j].block()`. Several methods on this
     /// struct rely on the fact that this ordering is observed.
     rows: Vec<EntityModification>,
 
-    immutable: bool,
-
     /// Map the `key.entity_id` of all entries in `rows` to the index with
     /// the most recent entry for that id to speed up lookups.
     last_mod: LastMod,
 }
 
 impl RowGroup {
-    pub fn new(entity_type: EntityType, immutable: bool) -> Self {
+    pub fn new(entity_type: EntityType) -> Self {
         Self {
             entity_type,
             rows: Vec::new(),
-            immutable,
             last_mod: LastMod::new(),
         }
     }
@@ -471,7 +471,7 @@ impl RowGroup {
     /// Append `row` to `self.rows` by combining it with a previously
     /// existing row, if that is possible
     fn append_row(&mut self, row: EntityModification) -> Result<(), StoreError> {
-        if self.immutable {
+        if self.entity_type.is_immutable() {
             match row {
                 EntityModification::Insert { .. } => {
                     // Check if this is an attempt to overwrite an immutable
@@ -488,6 +488,9 @@ impl RowGroup {
                         .and_then(|&idx| self.rows.get(idx))
                     {
                         Some(prev) if prev.block() != row.block() => {
+                            if self.entity_type.skip_duplicates() {
+                                return Ok(());
+                            }
                             return Err(StoreError::Input(
                                 format!("entity {} is immutable; inserting it at block {} is not possible as it was already inserted at block {}",
                                         row.key(), row.block(), prev.block())));
@@ -498,6 +501,9 @@ impl RowGroup {
                     self.push_row(row);
                 }
                 EntityModification::Overwrite { .. } | EntityModification::Remove { .. } => {
+                    if self.entity_type.skip_duplicates() {
+                        return Ok(());
+                    }
                     return Err(internal_error!(
                         "immutable entity type {} only allows inserts, not {:?}",
                         self.entity_type,
@@ -604,8 +610,8 @@ impl RowGroup {
 pub struct RowGroupForPerfTest(RowGroup);
 
 impl RowGroupForPerfTest {
-    pub fn new(entity_type: EntityType, immutable: bool) -> Self {
-        Self(RowGroup::new(entity_type, immutable))
+    pub fn new(entity_type: EntityType) -> Self {
+        Self(RowGroup::new(entity_type))
     }
 
     pub fn push(&mut self, emod: EntityModification, block: BlockNumber) -> Result<(), StoreError> {
@@ -661,11 +667,15 @@ impl<'a> Iterator for ClampsByBlockIterator<'a> {
 #[derive(Debug, CacheWeight)]
 pub struct RowGroups {
     pub groups: Vec<RowGroup>,
+    logger: Logger,
 }
 
 impl RowGroups {
-    fn new() -> Self {
-        Self { groups: Vec::new() }
+    fn new(logger: Logger) -> Self {
+        Self {
+            groups: Vec::new(),
+            logger,
+        }
     }
 
     fn group(&self, entity_type: &EntityType) -> Option<&RowGroup> {
@@ -684,9 +694,7 @@ impl RowGroups {
         match pos {
             Some(pos) => &mut self.groups[pos],
             None => {
-                let immutable = entity_type.is_immutable();
-                self.groups
-                    .push(RowGroup::new(entity_type.clone(), immutable));
+                self.groups.push(RowGroup::new(entity_type.clone()));
                 // unwrap: we just pushed an entry
                 self.groups.last_mut().unwrap()
             }
@@ -784,6 +792,7 @@ impl Batch {
         deterministic_errors: Vec<SubgraphError>,
         offchain_to_remove: Vec<StoredDynamicDataSource>,
         is_non_fatal_errors_active: bool,
+        logger: Logger,
     ) -> Result<Self, StoreError> {
         let block = block_ptr.number;
 
@@ -797,7 +806,7 @@ impl Batch {
             EntityModification::Remove { .. } => 0,
         });
 
-        let mut mods = RowGroups::new();
+        let mut mods = RowGroups::new(logger);
 
         for m in raw_mods {
             mods.group_entry(&m.key().entity_type).push(m, block)?;
@@ -1079,7 +1088,6 @@ mod test {
         let group = RowGroup {
             entity_type: ENTRY_TYPE.clone(),
             rows,
-            immutable: false,
             last_mod,
         };
         let act = group
@@ -1120,6 +1128,8 @@ mod test {
       type Thing @entity { id: ID!, count: Int! }
       type RowGroup @entity { id: ID! }
       type Entry @entity { id: ID! }
+      type ImmThing @entity(immutable: true) { id: ID!, count: Int! }
+      type SkipDupThing @entity(immutable: true, skipDuplicates: true) { id: ID!, count: Int! }
     "#;
     lazy_static! {
         static ref DEPLOYMENT: DeploymentHash = DeploymentHash::new("batchAppend").unwrap();
@@ -1128,6 +1138,8 @@ mod test {
         static ref THING_TYPE: EntityType = SCHEMA.entity_type("Thing").unwrap();
         static ref ROW_GROUP_TYPE: EntityType = SCHEMA.entity_type("RowGroup").unwrap();
         static ref ENTRY_TYPE: EntityType = SCHEMA.entity_type("Entry").unwrap();
+        static ref IMM_THING_TYPE: EntityType = SCHEMA.entity_type("ImmThing").unwrap();
+        static ref SKIP_DUP_THING_TYPE: EntityType = SCHEMA.entity_type("SkipDupThing").unwrap();
     }
 
     /// Convenient notation for changes to a fixed entity
@@ -1187,7 +1199,7 @@ mod test {
     impl Group {
         fn new() -> Self {
             Self {
-                group: RowGroup::new(THING_TYPE.clone(), false),
+                group: RowGroup::new(THING_TYPE.clone()),
             }
         }
 
@@ -1292,4 +1304,76 @@ mod test {
         let op = group.last_op(&key, 0);
         assert_eq!(None, op);
     }
+
+    fn make_insert(entity_type: &EntityType, id: &str, block: BlockNumber) -> EntityModification {
+        EntityModification::Insert {
+            key: entity_type.parse_key(id).unwrap(),
+            data: Arc::new(entity! { SCHEMA => id: id, count: block }),
+            block,
+            end: None,
+        }
+    }
+
+    fn make_overwrite(
+        entity_type: &EntityType,
+        id: &str,
+        block: BlockNumber,
+    ) -> EntityModification {
+        EntityModification::Overwrite {
+            key: entity_type.parse_key(id).unwrap(),
+            data: Arc::new(entity! { SCHEMA => id: id, count: block }),
+            block,
+            end: None,
+        }
+    }
+
+    fn make_remove(entity_type: &EntityType, id: &str, block: BlockNumber) -> EntityModification {
+        EntityModification::Remove {
+            key: entity_type.parse_key(id).unwrap(),
+            block,
+        }
+    }
+
+    #[test]
+    fn append_row_immutable_default_rejects_cross_block_duplicate() {
+        let mut group = RowGroup::new(IMM_THING_TYPE.clone());
+        let res = group.push(make_insert(&IMM_THING_TYPE, "one", 1), 1);
+        assert!(res.is_ok());
+        let res = group.push(make_insert(&IMM_THING_TYPE, "one", 2), 2);
+        assert!(res.is_err());
+    }
+
+    #[test]
+    fn append_row_skip_duplicates_allows_cross_block_duplicate() {
+        let mut group = RowGroup::new(SKIP_DUP_THING_TYPE.clone());
+        let res = group.push(make_insert(&SKIP_DUP_THING_TYPE, "one", 1), 1);
+        assert!(res.is_ok());
+        let res = group.push(make_insert(&SKIP_DUP_THING_TYPE, "one", 2), 2);
+        assert!(res.is_ok());
+        assert_eq!(group.row_count(), 1);
+    }
+
+    #[test]
+    fn append_row_skip_duplicates_allows_overwrite() {
+        let mut group = RowGroup::new(SKIP_DUP_THING_TYPE.clone());
+        let res = group.append_row(make_overwrite(&SKIP_DUP_THING_TYPE, "one", 1));
+        assert!(res.is_ok());
+    }
+
+    #[test]
+    fn append_row_skip_duplicates_allows_remove() {
+        let mut group = RowGroup::new(SKIP_DUP_THING_TYPE.clone());
+        let res = group.append_row(make_remove(&SKIP_DUP_THING_TYPE, "one", 1));
+        assert!(res.is_ok());
+    }
+
+    #[test]
+    fn append_row_skip_duplicates_same_block_still_pushes() {
+        let mut group = RowGroup::new(SKIP_DUP_THING_TYPE.clone());
+        let res = group.push(make_insert(&SKIP_DUP_THING_TYPE, "one", 1), 1);
+        assert!(res.is_ok());
+        let res = group.push(make_insert(&SKIP_DUP_THING_TYPE, "one", 1), 1);
+        assert!(res.is_ok());
+        assert_eq!(group.row_count(), 2);
+    }
 }

graph/src/schema/entity_type.rs

@@ -68,6 +68,10 @@ impl EntityType {
         self.schema.is_immutable(self.atom)
     }
 
+    pub fn skip_duplicates(&self) -> bool {
+        self.schema.skip_duplicates(self.atom)
+    }
+
     pub fn id_type(&self) -> Result<IdType, Error> {
         self.schema.id_type(self.atom)
     }