diff --git a/README.md b/README.md
index cd2178a..f3d5285 100644
--- a/README.md
+++ b/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:
diff --git a/ic_python_db/db_engine.py b/ic_python_db/db_engine.py
index e0a4b80..cd926f5 100644
--- a/ic_python_db/db_engine.py
+++ b/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.
diff --git a/ic_python_db/entity.py b/ic_python_db/entity.py
index 0ef43a2..b651968 100644
--- a/ic_python_db/entity.py
+++ b/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
diff --git a/ic_python_db/properties.py b/ic_python_db/properties.py
index ae6c135..153b5fb 100644
--- a/ic_python_db/properties.py
+++ b/ic_python_db/properties.py
@@ -30,12 +30,18 @@ class Property(Generic[T]):
     """Definition of an entity property.
 
     A generic descriptor class that provides type-safe property access.
+
+    With ``indexed=True`` a persistent secondary index (issue #11) maps each
+    distinct value to the list of entity IDs holding it, enabling equality
+    lookups via ``Entity.find_by`` without scanning all entities. ``None``
+    values are not indexed.
     """
 
     name: str
     type: Type[T]
     default: Optional[T]
     validator: Optional[Callable[[T], bool]]
+    indexed: bool
 
     def __init__(
         self,
@@ -43,11 +49,13 @@ def __init__(
         type: Type[T] = type(None),  # type: ignore[assignment]
         default: Optional[T] = None,
         validator: Optional[Callable[[T], bool]] = None,
+        indexed: bool = False,
     ):
         self.name = name
         self.type = type
         self.default = default
         self.validator = validator
+        self.indexed = indexed
 
     def __set_name__(self, owner: type, name: str) -> None:
         self.name = name
@@ -63,6 +71,11 @@ def __set__(self, obj, value):
         from .constants import ACTION_CREATE, ACTION_MODIFY
         from .hooks import call_entity_hook
 
+        # Fast path: hydrating from DB — store raw value, indexes are intact.
+        if getattr(obj, "_hydrating", False):
+            obj.__dict__[f"_{PROPERTY_STORAGE_PREFIX}_{self.name}"] = value
+            return
+
         old_value = obj.__dict__.get(
             f"_{PROPERTY_STORAGE_PREFIX}_{self.name}", self.default
         )
@@ -100,6 +113,14 @@ def __set__(self, obj, value):
                 raise ValueError(f"Invalid value for {self.name}: {value}")
 
         obj.__dict__[f"_{PROPERTY_STORAGE_PREFIX}_{self.name}"] = value
+
+        if self.indexed and old_value != value and obj._id is not None:
+            db = obj.db()
+            if old_value is not None:
+                db.field_index_remove(obj._type, self.name, str(old_value), obj._id)
+            if value is not None:
+                db.field_index_add(obj._type, self.name, str(value), obj._id)
+
         obj._save()
 
 
@@ -111,6 +132,7 @@ def __init__(
         min_length: Optional[int] = None,
         max_length: Optional[int] = None,
         default: Optional[str] = None,
+        indexed: bool = False,
     ):
         def validator(value: str) -> bool:
             if min_length is not None and len(value) < min_length:
@@ -119,7 +141,9 @@ def validator(value: str) -> bool:
                 return False
             return True
 
-        super().__init__(name="", type=str, default=default, validator=validator)
+        super().__init__(
+            name="", type=str, default=default, validator=validator, indexed=indexed
+        )
 
 
 class Integer(Property[int]):
@@ -130,6 +154,7 @@ def __init__(
         min_value: Optional[int] = None,
         max_value: Optional[int] = None,
         default: Optional[int] = None,
+        indexed: bool = False,
     ):
         def validator(value: int) -> bool:
             if min_value is not None and value < min_value:
@@ -138,7 +163,9 @@ def validator(value: int) -> bool:
                 return False
             return True
 
-        super().__init__(name="", type=int, default=default, validator=validator)
+        super().__init__(
+            name="", type=int, default=default, validator=validator, indexed=indexed
+        )
 
 
 class Float(Property[float]):
@@ -149,6 +176,7 @@ def __init__(
         min_value: Optional[float] = None,
         max_value: Optional[float] = None,
         default: Optional[float] = None,
+        indexed: bool = False,
     ):
         def validator(value: float) -> bool:
             if min_value is not None and value < min_value:
@@ -157,14 +185,16 @@ def validator(value: float) -> bool:
                 return False
             return True
 
-        super().__init__(name="", type=float, default=default, validator=validator)
+        super().__init__(
+            name="", type=float, default=default, validator=validator, indexed=indexed
+        )
 
 
 class Boolean(Property[bool]):
     """Boolean property."""
 
-    def __init__(self, default: Optional[bool] = None):
-        super().__init__(name="", type=bool, default=default)
+    def __init__(self, default: Optional[bool] = None, indexed: bool = False):
+        super().__init__(name="", type=bool, default=default, indexed=indexed)
 
 
 # ── Relation Properties ───────────────────────────────────────────────────────
diff --git a/tests/entrypoint.sh b/tests/entrypoint.sh
index 9f2a4f8..282029e 100755
--- a/tests/entrypoint.sh
+++ b/tests/entrypoint.sh
@@ -20,7 +20,7 @@ if [ ! -f "$TEMPLATE_PATH" ]; then
 fi
 
 # Define a list of test identifiers
-TEST_IDS=('example_1' 'example_2' 'entity' 'mixins' 'properties' 'alias_and_properties' 'relationships' 'enhanced_relations' 'serialization' 'database' 'audit' 'namespaces')
+TEST_IDS=('example_1' 'example_2' 'entity' 'mixins' 'properties' 'alias_and_properties' 'relationships' 'enhanced_relations' 'serialization' 'database' 'audit' 'namespaces' 'indexed_fields')
 
 if [ $# -gt 0 ]; then
   TEST_IDS=("$@")
diff --git a/tests/run_test.sh b/tests/run_test.sh
index 8c05932..bcc2c08 100755
--- a/tests/run_test.sh
+++ b/tests/run_test.sh
@@ -8,7 +8,7 @@ cd src
 
 exit_code=0
 
-TEST_IDS=("example_1" "example_2" "entity" "mixins" "properties" "alias_and_properties" "relationships" "enhanced_relations" "serialization" "namespaces" "migrations" "database" "audit" "auto_register" "schema")
+TEST_IDS=("example_1" "example_2" "entity" "mixins" "properties" "alias_and_properties" "relationships" "enhanced_relations" "serialization" "namespaces" "migrations" "database" "audit" "auto_register" "schema" "indexed_fields")
 
 # Check if a specific test ID is provided as an argument
 if [ "$1" ]; then
diff --git a/tests/src/main.py b/tests/src/main.py
index 7821b6d..8bdbdcc 100644
--- a/tests/src/main.py
+++ b/tests/src/main.py
@@ -9,6 +9,7 @@
     test_entity,
     test_example_1,
     test_example_2,
+    test_indexed_fields,
     test_mixins,
     test_namespaces,
     test_properties,
diff --git a/tests/src/tests/test_indexed_fields.py b/tests/src/tests/test_indexed_fields.py
new file mode 100644
index 0000000..f0f6118
--- /dev/null
+++ b/tests/src/tests/test_indexed_fields.py
@@ -0,0 +1,194 @@
+"""Tests for secondary field indexes (indexed=True, issue #11)."""
+
+from tester import Tester
+
+from ic_python_db import *
+
+
+class Ticket(Entity):
+    """Entity with a mix of indexed and non-indexed properties."""
+
+    title = String(max_length=100)
+    status = String(max_length=32, indexed=True)
+    priority = Integer(indexed=True)
+    is_open = Boolean(default=True, indexed=True)
+
+
+class TestIndexedFields:
+    def setUp(self):
+        Database.get_instance().clear()
+
+    # ── Basic lookups ────────────────────────────────────────────────────────
+
+    def test_find_by_returns_matches(self):
+        t1 = Ticket(title="A", status="open")
+        t2 = Ticket(title="B", status="open")
+        Ticket(title="C", status="closed")
+
+        entities, next_from_id = Ticket.find_by("status", "open")
+        assert {e._id for e in entities} == {t1._id, t2._id}
+        assert next_from_id is None
+
+        entities, _ = Ticket.find_by("status", "closed")
+        assert len(entities) == 1
+        assert entities[0].title == "C"
+
+    def test_find_by_no_matches(self):
+        Ticket(title="A", status="open")
+        entities, next_from_id = Ticket.find_by("status", "nonexistent")
+        assert entities == []
+        assert next_from_id is None
+
+    def test_find_by_unindexed_field_raises(self):
+        Ticket(title="A", status="open")
+        Tester.assert_raises(ValueError, lambda: Ticket.find_by("title", "A"))
+        Tester.assert_raises(ValueError, lambda: Ticket.count_by("title", "A"))
+        Tester.assert_raises(ValueError, lambda: Ticket.rebuild_field_index("title"))
+
+    def test_count_by(self):
+        Ticket(title="A", status="open")
+        Ticket(title="B", status="open")
+        Ticket(title="C", status="closed")
+        assert Ticket.count_by("status", "open") == 2
+        assert Ticket.count_by("status", "closed") == 1
+        assert Ticket.count_by("status", "missing") == 0
+
+    def test_non_string_values(self):
+        t1 = Ticket(title="A", priority=1)
+        Ticket(title="B", priority=2)
+        t3 = Ticket(title="C", priority=1, is_open=False)
+
+        entities, _ = Ticket.find_by("priority", 1)
+        assert {e._id for e in entities} == {t1._id, t3._id}
+
+        entities, _ = Ticket.find_by("is_open", False)
+        assert [e._id for e in entities] == [t3._id]
+
+    # ── Index maintenance on update / delete ────────────────────────────────
+
+    def test_update_moves_entity_between_values(self):
+        t = Ticket(title="A", status="open")
+        Ticket(title="B", status="open")
+
+        t.status = "closed"
+
+        open_entities, _ = Ticket.find_by("status", "open")
+        assert t._id not in {e._id for e in open_entities}
+        assert len(open_entities) == 1
+
+        closed_entities, _ = Ticket.find_by("status", "closed")
+        assert [e._id for e in closed_entities] == [t._id]
+
+    def test_set_to_none_removes_from_index(self):
+        t = Ticket(title="A", status="open")
+        t.status = None
+        assert Ticket.count_by("status", "open") == 0
+
+    def test_delete_removes_from_index(self):
+        t = Ticket(title="A", status="open")
+        Ticket(title="B", status="open")
+        t.delete()
+        entities, _ = Ticket.find_by("status", "open")
+        assert t._id not in {e._id for e in entities}
+        assert len(entities) == 1
+
+    def test_noop_update_keeps_index_consistent(self):
+        t = Ticket(title="A", status="open")
+        t.status = "open"  # same value
+        assert Ticket.count_by("status", "open") == 1
+
+    # ── Persistence across load ──────────────────────────────────────────────
+
+    def test_load_does_not_duplicate_or_corrupt_index(self):
+        t = Ticket(title="A", status="open")
+        entity_id = t._id
+        del t
+
+        db = Database.get_instance()
+        db.clear_registry()
+
+        loaded = Ticket.load(entity_id)
+        assert loaded.status == "open"
+        assert Ticket.count_by("status", "open") == 1
+
+        # Updating after a fresh load must still move the index entry.
+        loaded.status = "closed"
+        assert Ticket.count_by("status", "open") == 0
+        assert Ticket.count_by("status", "closed") == 1
+
+    # ── Pagination ────────────────────────────────────────────────────────────
+
+    def test_pagination_walks_all_matches_in_id_order(self):
+        ids = []
+        for i in range(7):
+            ids.append(Ticket(title=f"T{i}", status="open")._id)
+        Ticket(title="other", status="closed")
+
+        collected = []
+        cursor = 1
+        pages = 0
+        while cursor is not None:
+            entities, cursor = Ticket.find_by("status", "open", from_id=cursor, count=3)
+            collected.extend(e._id for e in entities)
+            pages += 1
+            assert pages < 10  # safety against infinite loop
+
+        assert collected == sorted(ids, key=int)
+        assert pages == 3  # 3 + 3 + 1
+
+    def test_pagination_from_id_skips_earlier_entities(self):
+        first = Ticket(title="A", status="open")
+        second = Ticket(title="B", status="open")
+
+        entities, _ = Ticket.find_by("status", "open", from_id=int(first._id) + 1)
+        assert [e._id for e in entities] == [second._id]
+
+    # ── Rebuild (backfill) ───────────────────────────────────────────────────
+
+    def test_rebuild_field_index_backfills_missing_entries(self):
+        t1 = Ticket(title="A", status="open")
+        t2 = Ticket(title="B", status="open")
+        t3 = Ticket(title="C", status="closed")
+
+        # Simulate entities written before the field was indexed: wipe all
+        # index entries for Ticket.status directly from storage.
+        db = Database.get_instance()
+        for key in list(db._db_storage.keys()):
+            if key.startswith("_fi:Ticket:status:"):
+                db._db_storage.remove(key)
+        assert Ticket.count_by("status", "open") == 0
+
+        cursor = 1
+        rounds = 0
+        while cursor is not None:
+            cursor = Ticket.rebuild_field_index("status", from_id=cursor, batch=2)
+            rounds += 1
+            assert rounds < 10
+
+        assert rounds == 2  # 3 entities, batch=2 → two rounds
+        assert Ticket.count_by("status", "open") == 2
+        assert Ticket.count_by("status", "closed") == 1
+
+        entities, _ = Ticket.find_by("status", "open")
+        assert {e._id for e in entities} == {t1._id, t2._id}
+        assert t3._id not in {e._id for e in entities}
+
+    def test_rebuild_is_idempotent(self):
+        Ticket(title="A", status="open")
+        cursor = 1
+        while cursor is not None:
+            cursor = Ticket.rebuild_field_index("status", from_id=cursor)
+        assert Ticket.count_by("status", "open") == 1
+
+    def test_rebuild_empty_type_returns_none(self):
+        assert Ticket.rebuild_field_index("status") is None
+
+
+def run(test_name: str = None, test_var: str = None):
+    Database.get_instance().clear()
+    tester = Tester(TestIndexedFields)
+    return tester.run_tests()
+
+
+if __name__ == "__main__":
+    exit(run())