feat: secondary indexes on scalar properties (indexed=True)

README.md

@@ -145,6 +145,42 @@ Then use the defined entities to store objects:
 
 For more usage examples, see the [tests](tests/src/tests).
 
+## Indexed Fields
+
+Declare `indexed=True` on a scalar property to maintain a persistent secondary index, enabling equality lookups without scanning all entities:
+
+```python
+class Proposal(Entity):
+    title = String()
+    status = String(max_length=32, indexed=True)
+    org_scope = String(max_length=128, indexed=True)
+
+Proposal(title="A", status="voting", org_scope="Justice")
+Proposal(title="B", status="executed", org_scope="Justice")
+
+# Paginated equality lookup — cost proportional to matches, not max_id
+entities, next_from_id = Proposal.find_by("status", "voting", from_id=1, count=50)
+
+# Count without loading entities
+Proposal.count_by("org_scope", "Justice")  # 2
+```
+
+The index is maintained automatically on create, update, and delete. `None` values are not indexed. Querying a field that is not declared `indexed=True` raises `ValueError`.
+
+### Backfilling existing data
+
+Entities created before a field became indexed are missing from its index. Rebuild in bounded batches (resumable, safe for IC instruction limits — spread calls across multiple update calls or timers):
+
+```python
+cursor = 1
+while cursor is not None:
+    cursor = Proposal.rebuild_field_index("status", from_id=cursor, batch=200)
+```
+
+The rebuild is idempotent. The `indexed` flag is not part of the schema descriptor, so toggling it never triggers upgrade-compatibility errors.
+
+Note: each distinct value's ID list is stored in a single storage entry, so extremely hot values are bounded by the backing store's `max_value_size` — the same constraint as relationship reverse indexes.
+
 ## Namespaces
 
 Organize entities with the `__namespace__` attribute to avoid type conflicts when you have the same class name in different modules:

ic_python_db/db_engine.py

@@ -426,6 +426,56 @@ def reverse_index_delete(
         if self._db_storage.get(key) is not None:
             self._db_storage.remove(key)
 
+    # ── Field Index API (issue #11) ────────────────────────────────────────────
+    #
+    # Secondary indexes for scalar properties declared with indexed=True.
+    # Allow equality lookups (Entity.find_by) without scanning all entities.
+    #
+    # Key format: _fi:{Type}:{field}:{value}
+    # Value: JSON array of entity IDs, e.g. ["1", "2", "3"]
+
+    def _fi_key(self, type_name: str, field_name: str, value: str) -> str:
+        return f"_fi:{type_name}:{field_name}:{value}"
+
+    def field_index_get(self, type_name: str, field_name: str, value: str) -> List[str]:
+        """Read the ID list for entities whose ``field_name`` equals ``value``.
+
+        Returns:
+            List of entity IDs, or empty list if no index entry exists.
+        """
+        key = self._fi_key(type_name, field_name, value)
+        raw = self._db_storage.get(key)
+        if raw:
+            return json.loads(raw)
+        return []
+
+    def field_index_add(
+        self, type_name: str, field_name: str, value: str, entity_id: str
+    ) -> None:
+        """Add an entity ID to the index entry for ``value``."""
+        key = self._fi_key(type_name, field_name, value)
+        raw = self._db_storage.get(key)
+        ids = json.loads(raw) if raw else []
+        if entity_id not in ids:
+            ids.append(entity_id)
+            self._db_storage.insert(key, json.dumps(ids))
+
+    def field_index_remove(
+        self, type_name: str, field_name: str, value: str, entity_id: str
+    ) -> None:
+        """Remove an entity ID from the index entry for ``value``."""
+        key = self._fi_key(type_name, field_name, value)
+        raw = self._db_storage.get(key)
+        if not raw:
+            return
+        ids = json.loads(raw)
+        if entity_id in ids:
+            ids.remove(entity_id)
+            if ids:
+                self._db_storage.insert(key, json.dumps(ids))
+            else:
+                self._db_storage.remove(key)
+
     # Reverse counters (unidirectional relations)
     # Key format: "_rc:{parent_type}:{parent_id}:{relation_name}"
     # Value: integer count of related entities.

ic_python_db/entity.py

@@ -1,6 +1,6 @@
 """Enhanced entity implementation with support for mixins and entity types."""
 
-from typing import Any, Dict, List, Optional, Set, Type, TypeVar
+from typing import Any, Dict, List, Optional, Set, Tuple, Type, TypeVar
 
 from ic_python_logging import get_logger
 
@@ -198,11 +198,17 @@ class User(Entity):
         self.db().register_entity(self)
 
         self._do_not_save = True
+        # True only while reconstructing persisted state in this constructor:
+        # property descriptors skip hooks, validation, and index maintenance
+        # (indexes are already intact). Explicit updates later — including
+        # deserialize() upserts — run the full descriptor path.
+        self._hydrating = self._loaded
         # Set additional attributes
         for k, v in kwargs.items():
             if not k.startswith("_"):
                 setattr(self, k, v)
         self._do_not_save = False
+        self._hydrating = False
 
         if not self._loaded:
             self._save()
@@ -454,6 +460,142 @@ def find(cls: Type[T], d) -> List[T]:
             if all(getattr(e, k, None) == v for k, v in d.items())
         ]
 
+    @classmethod
+    def _indexed_properties(cls) -> Dict[str, Any]:
+        """Return {name: Property} for all properties declared with indexed=True."""
+        from .properties import Property
+
+        result: Dict[str, Any] = {}
+        for klass in reversed(cls.__mro__):
+            for attr_name, attr_value in klass.__dict__.items():
+                if attr_name.startswith("_"):
+                    continue
+                if isinstance(attr_value, Property) and attr_value.indexed:
+                    result[attr_name] = attr_value
+        return result
+
+    @classmethod
+    def _require_indexed_field(cls, field: str) -> None:
+        if field not in cls._indexed_properties():
+            raise ValueError(
+                f"{cls.__name__}.{field} is not an indexed property "
+                f"(declare it with indexed=True)"
+            )
+
+    @classmethod
+    def find_by(
+        cls: Type[T],
+        field: str,
+        value: Any,
+        from_id: int = 1,
+        count: int = 50,
+    ) -> Tuple[List[T], Optional[int]]:
+        """Equality lookup on an indexed field with cursor pagination (issue #11).
+
+        IDs come from the persistent field index, so cost is proportional to
+        the number of matches, not to max_id. Results are ordered by ascending
+        entity ID, compatible with load_some-style cursors.
+
+        Args:
+            field: Name of a property declared with indexed=True
+            value: Value to match (compared via its string form, like the index)
+            from_id: Only return entities with _id >= from_id
+            count: Maximum number of entities to return
+
+        Returns:
+            (entities, next_from_id) — next_from_id is None when no more
+            matches remain, otherwise pass it back as from_id for the next page.
+
+        Raises:
+            ValueError: If the field is not declared with indexed=True
+        """
+        cls._require_indexed_field(field)
+        if from_id < 1:
+            raise ValueError("from_id must be at least 1")
+        if count < 1:
+            raise ValueError("count must be at least 1")
+
+        ids = cls.db().field_index_get(cls.get_full_type_name(), field, str(value))
+        matching = sorted((int(i) for i in ids if i.isdigit() and int(i) >= from_id))
+
+        entities: List[T] = []
+        last_id = None
+        for entity_id in matching:
+            if len(entities) >= count:
+                break
+            entity = cls.load(str(entity_id))
+            if entity is not None:
+                entities.append(entity)
+            last_id = entity_id
+
+        has_more = last_id is not None and any(i > last_id for i in matching)
+        next_from_id = (last_id + 1) if has_more else None
+        return entities, next_from_id
+
+    @classmethod
+    def count_by(cls: Type[T], field: str, value: Any) -> int:
+        """Number of entities whose indexed ``field`` equals ``value``.
+
+        Raises:
+            ValueError: If the field is not declared with indexed=True
+        """
+        cls._require_indexed_field(field)
+        return len(
+            cls.db().field_index_get(cls.get_full_type_name(), field, str(value))
+        )
+
+    @classmethod
+    def rebuild_field_index(
+        cls: Type[T],
+        field: str,
+        from_id: int = 1,
+        batch: int = 200,
+    ) -> Optional[int]:
+        """Index one batch of pre-existing entities; resumable (issue #11).
+
+        Entities created before a field was declared indexed=True are missing
+        from its index. This walks the ID range in bounded batches so callers
+        on the IC can spread the work across multiple update calls or timer
+        ticks without hitting per-message instruction limits:
+
+            cursor = 1
+            while cursor is not None:
+                cursor = Proposal.rebuild_field_index("org_scope", from_id=cursor)
+
+        Idempotent — re-indexing an already-indexed entity is a no-op.
+
+        Returns:
+            The from_id for the next batch, or None when the rebuild is done.
+
+        Raises:
+            ValueError: If the field is not declared with indexed=True
+        """
+        cls._require_indexed_field(field)
+        if from_id < 1:
+            raise ValueError("from_id must be at least 1")
+        if batch < 1:
+            raise ValueError("batch must be at least 1")
+
+        max_id = cls.max_id()
+        if max_id == 0 or from_id > max_id:
+            return None
+
+        db = cls.db()
+        type_name = cls.get_full_type_name()
+        end = min(from_id + batch - 1, max_id)
+        for entity_id in range(from_id, end + 1):
+            try:
+                entity = cls.load(str(entity_id))
+            except (ValueError, AttributeError):
+                continue
+            if entity is None:
+                continue
+            value = getattr(entity, field, None)
+            if value is not None:
+                db.field_index_add(type_name, field, str(value), entity._id)
+
+        return end + 1 if end < max_id else None
+
     @classmethod
     def instances(cls: Type[T]) -> List[T]:
         """Get all instances of this entity type, including subclass instances.
@@ -689,6 +831,12 @@ def delete(self) -> None:
                                 self._type, self._id, attr_value.reverse_name
                             )
 
+        # Remove from field indexes (issue #11)
+        for field_name in self._indexed_properties():
+            value = getattr(self, field_name, None)
+            if value is not None:
+                db.field_index_remove(self._type, field_name, str(value), self._id)
+
         db.delete(self._type, self._id)
 
         # Remove from entity registry