diff --git a/src/strawchemy/__init__.py b/src/strawchemy/__init__.py
index f9e25394..284d4cae 100644
--- a/src/strawchemy/__init__.py
+++ b/src/strawchemy/__init__.py
@@ -7,6 +7,17 @@
 from strawchemy.instance import ModelInstance
 from strawchemy.mapper import Strawchemy
 from strawchemy.repository.strawberry import StrawchemyAsyncRepository, StrawchemySyncRepository
+from strawchemy.schema.filters import (
+    ArrayComparison,
+    DateComparison,
+    DateTimeComparison,
+    EqualityComparison,
+    GraphQLComparison,
+    OrderComparison,
+    TextComparison,
+    TimeComparison,
+    TimeDeltaComparison,
+)
 from strawchemy.schema.interfaces import ErrorType
 from strawchemy.schema.mutation import (
     Input,
@@ -24,11 +35,17 @@
     "ALL",
     "RELATIONSHIPS",
     "SCALARS",
+    "ArrayComparison",
+    "DateComparison",
+    "DateTimeComparison",
+    "EqualityComparison",
     "ErrorType",
     "FieldGroup",
+    "GraphQLComparison",
     "Input",
     "InputValidationError",
     "ModelInstance",
+    "OrderComparison",
     "QueryHook",
     "RequiredToManyUpdateInput",
     "RequiredToOneInput",
@@ -36,6 +53,9 @@
     "StrawchemyAsyncRepository",
     "StrawchemyConfig",
     "StrawchemySyncRepository",
+    "TextComparison",
+    "TimeComparison",
+    "TimeDeltaComparison",
     "ToManyCreateInput",
     "ToManyUpdateInput",
     "ToOneInput",
diff --git a/src/strawchemy/dto/backend/strawberry.py b/src/strawchemy/dto/backend/strawberry.py
index 118007ea..af6e3e7d 100644
--- a/src/strawchemy/dto/backend/strawberry.py
+++ b/src/strawchemy/dto/backend/strawberry.py
@@ -50,6 +50,13 @@ def __init__(self, dto_base: type[AnnotatedDTOT], auto_is_type_of: bool = False)
         }
 
     def _construct_field_info(self, field_def: DTOFieldDefinition[ModelT, ModelFieldT]) -> FieldInfo:
+        # A field definition may carry a prebuilt strawberry field (e.g. from ``filter_field``);
+        # it already encodes the default and any explicit field config, so use it as-is.
+        # Local import: dto.strawberry imports this module at top, so a module-level import cycles.
+        from strawchemy.dto.strawberry import GraphQLFieldDefinition  # noqa: PLC0415
+
+        if isinstance(field_def, GraphQLFieldDefinition) and field_def.graphql_field is not None:
+            return FieldInfo(field_def.name, field_def.type_, field_def.graphql_field)
         strawberry_field: StrawberryField | None = None
         if field_def.default_factory is not DTOMissing:
             if isinstance(field_def.default_factory(), (list, tuple)):
diff --git a/src/strawchemy/dto/inspectors/sqlalchemy.py b/src/strawchemy/dto/inspectors/sqlalchemy.py
index 013f3bca..9912e9f2 100644
--- a/src/strawchemy/dto/inspectors/sqlalchemy.py
+++ b/src/strawchemy/dto/inspectors/sqlalchemy.py
@@ -61,7 +61,7 @@
     make_full_json_comparison_input,
     make_sqlite_json_comparison_input,
 )
-from strawchemy.utils.annotation import is_type_hint_optional
+from strawchemy.utils.annotation import get_origin_or_self, is_type_hint_optional
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Generator, Iterable
@@ -665,8 +665,8 @@ def _filter_type(cls, type_: type[Any], sqlalchemy_filter: type[GraphQLCompariso
         """
         return sqlalchemy_filter if cls._is_specialized(sqlalchemy_filter) else sqlalchemy_filter[type_]  # ty: ignore[not-subscriptable]  # runtime generic specialization with a dynamic type argument
 
-    def get_field_comparison(
-        self, field_definition: DTOFieldDefinition[DeclarativeBase, QueryableAttribute[Any]]
+    def get_comparison(
+        self, field_definition: DTOFieldDefinition[DeclarativeBase, QueryableAttribute[Any]], subscribed: bool = True
     ) -> type[GraphQLComparison]:
         """Determines the GraphQL comparison filter type for a DTO field.
 
@@ -678,14 +678,18 @@ def get_field_comparison(
         Args:
             field_definition: The DTO field definition, which contains information
                 about the model attribute and its type.
+            subscribed: When `False`, return the unsubscripted comparison class
+                (e.g. `OrderComparison` instead of `OrderComparison[int]`).
 
         Returns:
             The GraphQL comparison filter type suitable for the field.
         """
         field_type = field_definition.model_field.type
         if isinstance(field_type, ARRAY) and self.db_features.dialect == "postgresql":
-            return ArrayComparison[field_type.item_type.python_type]
-        return self.get_type_comparison(self.model_field_type(field_definition))
+            comparison: type[GraphQLComparison] = ArrayComparison[field_type.item_type.python_type]
+        else:
+            comparison = self.get_type_comparison(self.model_field_type(field_definition))
+        return comparison if subscribed else get_origin_or_self(comparison)
 
     def get_type_comparison(self, type_: type[Any]) -> type[GraphQLComparison]:
         """Determines the GraphQL comparison filter type for a Python type.
diff --git a/src/strawchemy/dto/strawberry.py b/src/strawchemy/dto/strawberry.py
index 5cf6e053..a6f27bf6 100644
--- a/src/strawchemy/dto/strawberry.py
+++ b/src/strawchemy/dto/strawberry.py
@@ -63,8 +63,10 @@
     from collections.abc import Callable, Hashable, Iterator
 
     from sqlalchemy import ColumnElement
+    from strawberry.types.field import StrawberryField
 
     from strawchemy.schema.filters import EqualityComparison, GraphQLComparison
+    from strawchemy.schema.filters.fields import CustomFilterApply, JoinStrategy
 
 T = TypeVar("T")
 
@@ -182,6 +184,8 @@ class GraphQLFieldDefinition(DTOFieldDefinition[DeclarativeBase, QueryableAttrib
     is_aggregate: bool = False
     is_function: bool = False
     is_function_arg: bool = False
+    graphql_field: StrawberryField | None = None
+    """Prebuilt strawberry field carrying explicit field config."""
 
     _function: FunctionInfo | None = None
 
@@ -286,6 +290,16 @@ def __post_init__(self) -> None:
         self.is_function_arg = True
 
 
+@dataclass(eq=False, repr=False, kw_only=True)
+class CustomFilterFieldDefinition(GraphQLFieldDefinition):
+    """Field definition for a custom-apply virtual filter field."""
+
+    apply: CustomFilterApply
+    """The custom filter callable; always set for this field type."""
+    join: JoinStrategy = "exists"
+    """Fold-back strategy used by the transpiler (``"exists"`` or ``"in"``)."""
+
+
 @dataclass(eq=False)
 class QueryNode(Node[GraphQLFieldDefinition, QueryNodeMetadata]):
     node_metadata: NodeMetadata[QueryNodeMetadata] | None = dataclasses.field(
@@ -354,9 +368,23 @@ class AggregationFilter:
     distinct: bool | None = None
 
 
+@dataclass
+class CustomFilter:
+    """A set custom-apply filter value, ready to be folded into the query."""
+
+    apply: CustomFilterApply
+    """The custom filter callable (see ``CustomFilterApply``)."""
+    value: Any
+    """The scalar value supplied in the GraphQL query."""
+    join: JoinStrategy
+    """Fold-back strategy (``"exists"`` or ``"in"``)."""
+    field_node: QueryNodeType
+    """The model's query node, used to correlate the EXISTS/IN subquery."""
+
+
 @dataclass
 class Filter:
-    and_: list[Self | GraphQLComparison | AggregationFilter] = dataclasses.field(default_factory=list)
+    and_: list[Self | GraphQLComparison | AggregationFilter | CustomFilter] = dataclasses.field(default_factory=list)
     or_: list[Self] = dataclasses.field(default_factory=list)
     not_: Self | None = None
 
@@ -531,7 +559,9 @@ def filters_tree(self, _node: QueryNodeType | None = None) -> tuple[QueryNodeTyp
         for name in self.dto_set_fields:
             value: EqualityComparison[Any] | BooleanFilterDTO | AggregateFilterDTO = getattr(self, name)
             field = self.__dto_field_definitions__[name]
-            if isinstance(value, BooleanFilterDTO):
+            if isinstance(field, CustomFilterFieldDefinition):
+                query.and_.append(CustomFilter(apply=field.apply, value=value, join=field.join, field_node=node))
+            elif isinstance(value, BooleanFilterDTO):
                 child, _ = node.upsert_child(field, match_on="value_equality")
                 _, sub_query = value.filters_tree(child)
                 if sub_query:
diff --git a/src/strawchemy/mapper.py b/src/strawchemy/mapper.py
index ea791874..ab36363d 100644
--- a/src/strawchemy/mapper.py
+++ b/src/strawchemy/mapper.py
@@ -6,12 +6,14 @@
 
 from strawberry.annotation import StrawberryAnnotation
 from strawberry.schema.config import StrawberryConfig
+from typing_extensions import Unpack
 
 from strawchemy.config.base import StrawchemyConfig
 from strawchemy.dto.backend.strawberry import StrawberrryDTOBackend
 from strawchemy.dto.base import TYPING_NS
 from strawchemy.dto.strawberry import BooleanFilterDTO, EnumDTO, MappedStrawberryGraphQLDTO, OrderByDTO, OrderByEnum
 from strawchemy.dto.utils import read_all_config
+from strawchemy.exceptions import StrawchemyFieldError
 from strawchemy.schema.factories import (
     AggregateFilterFactory,
     AggregateRootTypeFactory,
@@ -25,7 +27,8 @@
     UpsertConflictEnumBackend,
     UpsertConflictEnumFactory,
 )
-from strawchemy.schema.field import StrawchemyField
+from strawchemy.schema.field import MutationFieldKwargs, OutputFieldKwargs, StrawberryFieldKwargs, StrawchemyField
+from strawchemy.schema.filters.fields import VALID_JOINS, FilterFieldMarker
 from strawchemy.schema.mutation import types as mutation_types
 from strawchemy.schema.mutation.field_builder import MutationFieldBuilder
 from strawchemy.schema.mutation.fields import (
@@ -38,15 +41,14 @@
 from strawchemy.utils.registry import StrawberryRegistry
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Mapping, Sequence
+    from collections.abc import Sequence
 
     from sqlalchemy.orm import DeclarativeBase
-    from strawberry import BasePermission
-    from strawberry.extensions.field_extension import FieldExtension
     from strawberry.types.arguments import StrawberryArgument
 
     from strawchemy.dto.types import FieldSpec
     from strawchemy.repository.typing import QueryHookCallable
+    from strawchemy.schema.filters.fields import CustomFilterApply, JoinStrategy
     from strawchemy.schema.pagination import DefaultOffsetPagination
     from strawchemy.transpiler.hook import QueryHook
     from strawchemy.typing import (
@@ -201,17 +203,8 @@ def field(
         execution_options: dict[str, Any] | None = None,
         query_hook: QueryHook[Any] | Sequence[QueryHook[Any]] | None = None,
         repository_type: AnyRepositoryType | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        graphql_type: Any | None = None,
-        extensions: list[FieldExtension] | None = None,
         root_field: bool = True,
+        **field_kwargs: Unpack[OutputFieldKwargs],
     ) -> StrawchemyField: ...
 
     @overload
@@ -231,17 +224,8 @@ def field(
         execution_options: dict[str, Any] | None = None,
         query_hook: QueryHookCallable[Any] | Sequence[QueryHookCallable[Any]] | None = None,
         repository_type: AnyRepositoryType | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        graphql_type: Any | None = None,
-        extensions: list[FieldExtension] | None = None,
         root_field: bool = True,
+        **field_kwargs: Unpack[OutputFieldKwargs],
     ) -> Any: ...
 
     def field(
@@ -261,17 +245,8 @@ def field(
         execution_options: dict[str, Any] | None = None,
         query_hook: QueryHookCallable[Any] | Sequence[QueryHookCallable[Any]] | None = None,
         repository_type: AnyRepositoryType | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        graphql_type: Any | None = None,
-        extensions: list[FieldExtension] | None = None,
         root_field: bool = True,
+        **field_kwargs: Unpack[OutputFieldKwargs],
     ) -> Any:
         """Creates a Strawberry GraphQL field with enhanced SQLAlchemy capabilities.
 
@@ -303,22 +278,15 @@ def field(
             execution_options: SQLAlchemy execution options for the query.
             query_hook: A callable or sequence of callables to modify the SQLAlchemy query.
             repository_type: A custom strawberry class for data fetching logic.
-            name: The name of the GraphQL field.
-            description: The description of the GraphQL field.
-            permission_classes: A list of permission classes for the field.
-            deprecation_reason: The reason for deprecating the field.
-            default: The default value for the field.
-            default_factory: A factory function to generate the default value.
-            metadata: Additional metadata for the field.
-            directives: A sequence of directives for the field.
-            graphql_type: The GraphQL type of the field. If not provided, it's inferred.
-            extensions: A list of Strawberry FieldExtensions.
             root_field: Indicates if this is a root-level field.
+            **field_kwargs: ``strawberry.field`` arguments forwarded to the generated field
+                (see ``OutputFieldKwargs``). ``name`` maps to the GraphQL field name.
 
         Returns:
             A StrawchemyField instance, which is a specialized StrawberryField.
         """
         namespace = self._annotation_namespace()
+        graphql_type = field_kwargs.get("graphql_type")
         type_annotation = StrawberryAnnotation.from_annotation(graphql_type, namespace) if graphql_type else None
 
         if model_field is not None:
@@ -340,18 +308,18 @@ def field(
             query_hook=query_hook,
             model_field=model_field,
             python_name=None,
-            graphql_name=name,
+            graphql_name=field_kwargs.get("name"),
             type_annotation=type_annotation,
             is_subscription=False,
-            permission_classes=permission_classes or [],
-            deprecation_reason=deprecation_reason,
-            default=default,
-            default_factory=default_factory,
-            metadata=metadata,
-            directives=directives,
-            extensions=extensions or [],
+            permission_classes=field_kwargs.get("permission_classes") or [],
+            deprecation_reason=field_kwargs.get("deprecation_reason"),
+            default=field_kwargs.get("default", dataclasses.MISSING),
+            default_factory=field_kwargs.get("default_factory", dataclasses.MISSING),
+            metadata=field_kwargs.get("metadata"),
+            directives=field_kwargs.get("directives", ()),
+            extensions=field_kwargs.get("extensions") or [],
             registry_namespace=namespace,
-            description=description,
+            description=field_kwargs.get("description"),
             arguments=arguments,
             order_by_factory=self.order_by_factory,
             filter_factory=self.filter_factory,
@@ -359,23 +327,53 @@ def field(
         )
         return field(resolver) if resolver else field
 
+    def filter_field(
+        self,
+        *,
+        ops: Sequence[str] | None = None,
+        apply: CustomFilterApply | None = None,
+        join: JoinStrategy = "exists",
+        **field_kwargs: Unpack[StrawberryFieldKwargs],
+    ) -> Any:
+        """Declares a fine-grained filter field default.
+
+        The field's annotation supplies the comparison data type. With ``ops`` the field exposes
+        only those GraphQL operators; with ``apply`` it becomes a custom virtual scalar input;
+        with neither it force-includes the column's full default comparison.
+
+        Args:
+            ops: GraphQL operator names to expose (restricted field). Mutually exclusive with ``apply``.
+            apply: Custom filter callable ``(statement, value, *, dialect, model)`` returning a
+                mutated ``Select``.
+            join: Fold-back strategy when ``apply`` is set (``"exists"`` or ``"in"``).
+            **field_kwargs: ``strawberry.field`` arguments applied to the generated GraphQL field
+                (``name``, ``description``, ``metadata``, ``deprecation_reason``, ``directives``,
+                ``graphql_type``).
+
+        Returns:
+            A ``FilterFieldMarker`` consumed by the filter factory. Typed ``Any`` so it can sit as
+            a default under any annotation.
+
+        Raises:
+            StrawchemyFieldError: If ``ops`` and ``apply`` are both given, or ``join`` is unsupported.
+        """
+        if ops is not None and apply is not None:
+            msg = "filter_field() arguments 'ops' and 'apply' are mutually exclusive"
+            raise StrawchemyFieldError(msg)
+        if join not in VALID_JOINS:
+            msg = f"Invalid join strategy {join!r}; expected one of {sorted(VALID_JOINS)}"
+            raise StrawchemyFieldError(msg)
+        return FilterFieldMarker(
+            ops=tuple(ops) if ops is not None else None, apply=apply, join=join, field_kwargs=field_kwargs
+        )
+
     def create(
         self,
         input_type: type[MappedGraphQLDTO[T]],
         resolver: Any | None = None,
         *,
-        repository_type: AnyRepositoryType | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        graphql_type: Any | None = None,
-        extensions: list[FieldExtension] | None = None,
         validation: ValidationProtocol[T] | None = None,
+        **field_kwargs: Unpack[MutationFieldKwargs],
     ) -> Any:
         """Creates a Strawberry GraphQL mutation field for creating new model instances.
 
@@ -389,22 +387,11 @@ def create(
                 a new model instance. This should be a `MappedGraphQLDTO`.
             resolver: An optional custom resolver function for the mutation. If not
                 provided, Strawchemy will use a default resolver.
-            repository_type: An optional custom strawberry class for data fetching
-                and persistence logic. Defaults to the strawberry configured in
-                `StrawchemyConfig`.
-            name: The name of the GraphQL mutation field.
-            description: The description of the GraphQL mutation field.
-            permission_classes: A list of permission classes for the field.
-            deprecation_reason: The reason for deprecating the field.
-            default: The default value for the field (typically not used for mutations).
-            default_factory: A factory function to generate the default value.
-            metadata: Additional metadata for the field.
-            directives: A sequence of directives for the field.
-            graphql_type: The GraphQL return type of the mutation. If not provided,
-                it's inferred, typically to be the corresponding output type of the model.
-            extensions: A list of Strawberry FieldExtensions.
             validation: An optional validation protocol instance to validate
                 the input data before creation.
+            **field_kwargs: Common ``strawberry.field`` / repository arguments
+                (see ``MutationFieldKwargs``); ``graphql_type`` sets the mutation
+                return type.
 
         Returns:
             A `StrawchemyCreateMutationField` instance, which is a specialized
@@ -413,19 +400,9 @@ def create(
         return self._mutation_builder.build(
             StrawchemyCreateMutationField,
             resolver,
-            repository_type=repository_type,
-            graphql_type=graphql_type,
-            name=name,
-            description=description,
-            permission_classes=permission_classes,
-            deprecation_reason=deprecation_reason,
-            default=default,
-            default_factory=default_factory,
-            metadata=metadata,
-            directives=directives,
-            extensions=extensions,
             input_type=input_type,
             validation=validation,
+            **field_kwargs,
         )
 
     def upsert(
@@ -435,18 +412,8 @@ def upsert(
         conflict_fields: type[EnumDTO],
         resolver: Any | None = None,
         *,
-        repository_type: AnyRepositoryType | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        graphql_type: Any | None = None,
-        extensions: list[FieldExtension] | None = None,
         validation: ValidationProtocol[T] | None = None,
+        **field_kwargs: Unpack[MutationFieldKwargs],
     ) -> Any:
         """Creates a Strawberry GraphQL mutation field for upserting model instances.
 
@@ -465,22 +432,11 @@ def upsert(
                 conflict detection (e.g., primary key or unique constraints).
             resolver: An optional custom resolver function for the mutation. If not
                 provided, Strawchemy will use a default resolver.
-            repository_type: An optional custom strawberry class for data fetching
-                and persistence logic. Defaults to the strawberry configured in
-                `StrawchemyConfig`.
-            name: The name of the GraphQL mutation field.
-            description: The description of the GraphQL mutation field.
-            permission_classes: A list of permission classes for the field.
-            deprecation_reason: The reason for deprecating the field.
-            default: The default value for the field (typically not used for mutations).
-            default_factory: A factory function to generate the default value.
-            metadata: Additional metadata for the field.
-            directives: A sequence of directives for the field.
-            graphql_type: The GraphQL return type of the mutation. If not provided,
-                it's inferred, typically to be the corresponding output type of the model.
-            extensions: A list of Strawberry FieldExtensions.
             validation: An optional validation protocol instance to validate
                 the input data before the upsert operation.
+            **field_kwargs: Common ``strawberry.field`` / repository arguments
+                (see ``MutationFieldKwargs``); ``graphql_type`` sets the mutation
+                return type.
 
         Returns:
             A `StrawchemyUpsertMutationField` instance, which is a specialized
@@ -489,21 +445,11 @@ def upsert(
         return self._mutation_builder.build(
             StrawchemyUpsertMutationField,
             resolver,
-            repository_type=repository_type,
-            graphql_type=graphql_type,
-            name=name,
-            description=description,
-            permission_classes=permission_classes,
-            deprecation_reason=deprecation_reason,
-            default=default,
-            default_factory=default_factory,
-            metadata=metadata,
-            directives=directives,
-            extensions=extensions,
             input_type=input_type,
             update_fields_enum=update_fields,
             conflict_fields_enum=conflict_fields,
             validation=validation,
+            **field_kwargs,
         )
 
     def update(
@@ -512,18 +458,8 @@ def update(
         filter_input: type[BooleanFilterDTO],
         resolver: Any | None = None,
         *,
-        repository_type: AnyRepositoryType | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        graphql_type: Any | None = None,
-        extensions: list[FieldExtension] | None = None,
         validation: ValidationProtocol[T] | None = None,
+        **field_kwargs: Unpack[MutationFieldKwargs],
     ) -> Any:
         """Creates a Strawberry GraphQL mutation field for updating model instances.
 
@@ -540,23 +476,11 @@ def update(
                 instances should be updated. This should be a `BooleanFilterDTO`.
             resolver: An optional custom resolver function for the mutation. If not
                 provided, Strawchemy will use a default resolver.
-            repository_type: An optional custom strawberry class for data fetching
-                and persistence logic. Defaults to the strawberry configured in
-                `StrawchemyConfig`.
-            name: The name of the GraphQL mutation field.
-            description: The description of the GraphQL mutation field.
-            permission_classes: A list of permission classes for the field.
-            deprecation_reason: The reason for deprecating the field.
-            default: The default value for the field (typically not used for mutations).
-            default_factory: A factory function to generate the default value.
-            metadata: Additional metadata for the field.
-            directives: A sequence of directives for the field.
-            graphql_type: The GraphQL return type of the mutation. If not provided,
-                it's inferred, typically to be a list of the corresponding output
-                type of the model or a success/failure indicator.
-            extensions: A list of Strawberry FieldExtensions.
             validation: An optional validation protocol instance to validate
                 the input data before the update operation.
+            **field_kwargs: Common ``strawberry.field`` / repository arguments
+                (see ``MutationFieldKwargs``); ``graphql_type`` sets the mutation
+                return type.
 
         Returns:
             A `StrawchemyUpdateMutationField` instance, which is a specialized
@@ -565,20 +489,10 @@ def update(
         return self._mutation_builder.build(
             StrawchemyUpdateMutationField,
             resolver,
-            repository_type=repository_type,
-            graphql_type=graphql_type,
-            name=name,
-            description=description,
-            permission_classes=permission_classes,
-            deprecation_reason=deprecation_reason,
-            default=default,
-            default_factory=default_factory,
-            metadata=metadata,
-            directives=directives,
-            extensions=extensions,
             input_type=input_type,
             filter_type=filter_input,
             validation=validation,
+            **field_kwargs,
         )
 
     def update_by_ids(
@@ -586,18 +500,8 @@ def update_by_ids(
         input_type: type[MappedGraphQLDTO[T]],
         resolver: Any | None = None,
         *,
-        repository_type: AnyRepositoryType | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        graphql_type: Any | None = None,
-        extensions: list[FieldExtension] | None = None,
         validation: ValidationProtocol[T] | None = None,
+        **field_kwargs: Unpack[MutationFieldKwargs],
     ) -> Any:
         """Creates a Strawberry GraphQL mutation field for updating model instances by IDs.
 
@@ -613,23 +517,11 @@ def update_by_ids(
                 generated by `pk_update_input`, which includes primary key fields.
             resolver: An optional custom resolver function for the mutation. If not
                 provided, Strawchemy will use a default resolver.
-            repository_type: An optional custom strawberry class for data fetching
-                and persistence logic. Defaults to the strawberry configured in
-                `StrawchemyConfig`.
-            name: The name of the GraphQL mutation field.
-            description: The description of the GraphQL mutation field.
-            permission_classes: A list of permission classes for the field.
-            deprecation_reason: The reason for deprecating the field.
-            default: The default value for the field (typically not used for mutations).
-            default_factory: A factory function to generate the default value.
-            metadata: Additional metadata for the field.
-            directives: A sequence of directives for the field.
-            graphql_type: The GraphQL return type of the mutation. If not provided,
-                it's inferred, typically to be the corresponding output type of the
-                model or a list thereof.
-            extensions: A list of Strawberry FieldExtensions.
             validation: An optional validation protocol instance to validate
                 the input data before the update operation.
+            **field_kwargs: Common ``strawberry.field`` / repository arguments
+                (see ``MutationFieldKwargs``); ``graphql_type`` sets the mutation
+                return type.
 
         Returns:
             A `StrawchemyUpdateMutationField` instance, specialized for updates
@@ -638,37 +530,16 @@ def update_by_ids(
         return self._mutation_builder.build(
             StrawchemyUpdateMutationField,
             resolver,
-            repository_type=repository_type,
-            graphql_type=graphql_type,
-            name=name,
-            description=description,
-            permission_classes=permission_classes,
-            deprecation_reason=deprecation_reason,
-            default=default,
-            default_factory=default_factory,
-            metadata=metadata,
-            directives=directives,
-            extensions=extensions,
             input_type=input_type,
             validation=validation,
+            **field_kwargs,
         )
 
     def delete(
         self,
         filter_input: type[BooleanFilterDTO] | None = None,
         resolver: Any | None = None,
-        *,
-        repository_type: AnyRepositoryType | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        graphql_type: Any | None = None,
-        extensions: list[FieldExtension] | None = None,
+        **field_kwargs: Unpack[MutationFieldKwargs],
     ) -> Any:
         """Creates a Strawberry GraphQL mutation field for deleting model instances.
 
@@ -686,21 +557,9 @@ def delete(
                 record based on an ID passed directly (implementation dependent).
             resolver: An optional custom resolver function for the mutation. If not
                 provided, Strawchemy will use a default resolver.
-            repository_type: An optional custom strawberry class for data fetching
-                and persistence logic. Defaults to the strawberry configured in
-                `StrawchemyConfig`.
-            name: The name of the GraphQL mutation field.
-            description: The description of the GraphQL mutation field.
-            permission_classes: A list of permission classes for the field.
-            deprecation_reason: The reason for deprecating the field.
-            default: The default value for the field (typically not used for mutations).
-            default_factory: A factory function to generate the default value.
-            metadata: Additional metadata for the field.
-            directives: A sequence of directives for the field.
-            graphql_type: The GraphQL return type of the mutation. If not provided,
-                it's inferred, often to indicate success/failure or the number
-                of records deleted.
-            extensions: A list of Strawberry FieldExtensions.
+            **field_kwargs: Common ``strawberry.field`` / repository arguments
+                (see ``MutationFieldKwargs``); ``graphql_type`` sets the mutation
+                return type.
 
         Returns:
             A `StrawchemyDeleteMutationField` instance, which is a specialized
@@ -709,16 +568,6 @@ def delete(
         return self._mutation_builder.build(
             StrawchemyDeleteMutationField,
             resolver,
-            repository_type=repository_type,
-            graphql_type=graphql_type,
-            name=name,
-            description=description,
-            permission_classes=permission_classes,
-            deprecation_reason=deprecation_reason,
-            default=default,
-            default_factory=default_factory,
-            metadata=metadata,
-            directives=directives,
-            extensions=extensions,
             input_type=filter_input,
+            **field_kwargs,
         )
diff --git a/src/strawchemy/schema/factories/inputs.py b/src/strawchemy/schema/factories/inputs.py
index b4815b9d..89defc4e 100644
--- a/src/strawchemy/schema/factories/inputs.py
+++ b/src/strawchemy/schema/factories/inputs.py
@@ -1,7 +1,9 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Literal, Optional, TypeVar, Union
+import inspect
+from typing import TYPE_CHECKING, Any, Literal, Optional, TypeVar, Union, cast
 
+import strawberry
 from strawberry import UNSET
 from typing_extensions import Unpack, override
 
@@ -11,6 +13,7 @@
     AggregateFilterDTO,
     AggregationFunctionFilterDTO,
     BooleanFilterDTO,
+    CustomFilterFieldDefinition,
     FilterFunctionInfo,
     FunctionArgFieldDefinition,
     FunctionFieldDefinition,
@@ -19,14 +22,19 @@
     OrderByEnum,
 )
 from strawchemy.dto.types import DTOConfig, DTOMissing, Purpose
+from strawchemy.exceptions import StrawchemyFieldError
 from strawchemy.schema.factories import AggregationInspector, StrawchemyUnMappedFactory, UnmappedGraphQLDTOT
+from strawchemy.schema.filters import GraphQLComparison
+from strawchemy.schema.filters.fields import FilterFieldMarker
 from strawchemy.typing import AggregationFunction, GraphQLFilterDTOT, GraphQLPurpose, GraphQLType
+from strawchemy.utils.annotation import get_origin_or_self
 from strawchemy.utils.text import snake_to_camel
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Generator
+    from collections.abc import Callable, Generator, Sequence
 
     from sqlalchemy.orm import DeclarativeBase, QueryableAttribute
+    from strawberry.types.field import StrawberryField
 
     from strawchemy import Strawchemy
     from strawchemy.dto.base import DTOBackend, DTOBase, DTOFieldDefinition, ModelFieldT, Relation
@@ -62,6 +70,33 @@ def input(
     ) -> Callable[[type[Any]], type[UnmappedGraphQLDTOT]]:
         return self._input_wrapper(model=model, name=name, purpose=purpose, mode=mode, **kwargs)
 
+    @staticmethod
+    def parse_declared_filter_fields(base: type[Any]) -> dict[str, tuple[FilterFieldMarker, Any]]:
+        """Extracts user-declared filter fields from a decorated filter class.
+
+        A declared field is a class attribute whose value is a ``FilterFieldMarker`` (from
+        ``strawchemy.filter_field()``). Its annotation supplies the comparison data type.
+
+        Args:
+            base: The decorated filter class.
+
+        Returns:
+            Mapping of field name to ``(marker, annotation_type)``.
+
+        Raises:
+            StrawchemyFieldError: If a restricted (``ops``) or custom (``apply``) field has no annotation.
+        """
+        # Resolve string annotations (modules using `from __future__ import annotations`).
+        annotations = inspect.get_annotations(base, eval_str=True)
+        declared: dict[str, tuple[FilterFieldMarker, Any]] = {}
+        for name, marker in inspect.getmembers(base, lambda v: isinstance(v, FilterFieldMarker)):
+            annotation = annotations.get(name)
+            if annotation is None and (marker.ops is not None or marker.apply is not None):
+                msg = f"Filter field {name!r} needs a type annotation to determine its data type"
+                raise StrawchemyFieldError(msg)
+            declared[name] = (marker, annotation)
+        return declared
+
 
 class _FilterFactory(_BaseFilterFactory[GraphQLFilterDTOT]):
     def __init__(
@@ -77,7 +112,7 @@ def __init__(
         self._aggregation_filter_factory = aggregation_filter_factory or AggregateFilterFactory(mapper)
 
     def _filter_type(self, field: DTOFieldDefinition[DeclarativeBase, QueryableAttribute[Any]]) -> type[GraphQLFilter]:
-        return self.inspector.get_field_comparison(field)
+        return self.inspector.get_comparison(field)
 
     def _aggregation_field(
         self, field_def: DTOFieldDefinition[DeclarativeBase, QueryableAttribute[Any]], dto_config: DTOConfig
@@ -95,6 +130,94 @@ def _aggregation_field(
             default=UNSET,
         )
 
+    @staticmethod
+    def _validate_declared_columns(
+        declared: dict[str, tuple[FilterFieldMarker, Any]], matched: set[str], model: type[Any]
+    ) -> None:
+        """Validates that non-custom declared filter fields map to real model columns.
+
+        Restricted (``ops``) and bare declared fields must name an actual column; relationship
+        targets and unknown names are out of scope and raise. Custom-apply fields (``apply`` set)
+        are virtual and skip this check.
+
+        Args:
+            declared: User-declared filter fields keyed by attribute name.
+            matched: Names of declared fields that matched a real model column.
+            model: The SQLAlchemy model the filter targets.
+
+        Raises:
+            StrawchemyFieldError: If a non-custom declared field does not map to a column.
+        """
+        for field_name, (marker, _annotation) in declared.items():
+            if marker.apply is None and field_name not in matched:
+                msg = f"Filter field {field_name!r} is not a column on {model.__name__}"
+                raise StrawchemyFieldError(msg)
+
+    def _restricted_comparison(
+        self, field: DTOFieldDefinition[DeclarativeBase, QueryableAttribute[Any]], annotation: Any, ops: Sequence[str]
+    ) -> type[GraphQLComparison]:
+        """Builds the operator-restricted comparison for a declared filter field.
+
+        The comparison shape is derived from the column; the field annotation is documentary and
+        may be the scalar type or a comparison type (e.g. ``TextComparison``) reflecting the shape.
+
+        Args:
+            field: The model column the filter targets.
+            annotation: The declared field annotation (scalar or comparison type).
+            ops: Selected GraphQL operator names.
+
+        Returns:
+            The restricted comparison input type for the column.
+
+        Raises:
+            StrawchemyFieldError: If a comparison-type annotation does not match the column.
+        """
+        comparison_cls = self.inspector.get_comparison(field, subscribed=False)
+        annotation_origin = get_origin_or_self(annotation)
+        if (
+            isinstance(annotation_origin, type)
+            and issubclass(annotation_origin, GraphQLComparison)
+            and annotation_origin is not comparison_cls
+        ):
+            msg = (
+                f"Filter field {field.model_field_name!r} is annotated with "
+                f"{annotation_origin.__name__} but its column resolves to {comparison_cls.__name__}"
+            )
+            raise StrawchemyFieldError(msg)
+        return comparison_cls.restricted(self.inspector.model_field_type(field), tuple(ops))
+
+    def _iter_custom_filter_fields(
+        self,
+        declared_filters: dict[str, tuple[FilterFieldMarker, Any]],
+        model: type[DeclarativeT],
+        dto_config: DTOConfig,
+    ) -> Generator[DTOFieldDefinition[DeclarativeBase, QueryableAttribute[Any]]]:
+        """Yields custom-apply virtual filter fields (those declared with ``apply``).
+
+        These are not backed by a model column; each becomes a ``CustomFilterFieldDefinition``
+        whose value the transpiler folds into the query via the callable.
+        """
+        for field_name, (declared_filter, annotation) in declared_filters.items():
+            if declared_filter.apply is None:
+                continue
+            custom_field = CustomFilterFieldDefinition(
+                dto_config=dto_config,
+                model=model,
+                model_field_name=field_name,
+                type_hint=Optional[annotation],
+                apply=declared_filter.apply,
+                join=declared_filter.join,
+                graphql_field=self._declared_strawberry_field(declared_filter),
+                default=UNSET,
+                default_factory=DTOMissing,
+            )
+            yield custom_field
+
+    @staticmethod
+    def _declared_strawberry_field(marker: FilterFieldMarker) -> StrawberryField:
+        """Builds the strawberry field for a declared filter field from its ``field_kwargs``."""
+        return strawberry.field(default=UNSET, **marker.field_kwargs)
+
     @override
     def iter_field_definitions(
         self,
@@ -108,6 +231,8 @@ def iter_field_definitions(
         aggregate_filters: bool = False,
         **kwargs: Any,
     ) -> Generator[DTOFieldDefinition[DeclarativeBase, QueryableAttribute[Any]]]:
+        declared_filters = self.parse_declared_filter_fields(base) if base is not None else {}
+        matched_fields: set[str] = set()  # restricted-op declared fields matched to a real column
         for field in super().iter_field_definitions(name, model, dto_config, base, node, if_no_fields, **kwargs):
             if field.is_relation:
                 field.type_ = Union[field.type_, None]
@@ -116,13 +241,47 @@ def iter_field_definitions(
                 if aggregate_filters:
                     yield self._aggregation_field(field, dto_config.copy_with(partial_default=UNSET, partial=True))
             else:
-                comparison_type = self._filter_type(field)
+                declared_entry = declared_filters.get(field.model_field_name)
+                declared_filter = declared_entry[0] if declared_entry is not None else None
+                annotation = declared_entry[1] if declared_entry is not None else None
+                if declared_filter is not None and declared_filter.apply is not None:
+                    # Custom-apply field overrides this real column; skip the default
+                    # comparison and let the injection loop below emit the custom field once.
+                    continue
+                if declared_filter is not None and declared_filter.ops is not None:
+                    comparison_type = self._restricted_comparison(field, annotation, declared_filter.ops)
+                    matched_fields.add(field.model_field_name)
+                elif declared_filter is not None:
+                    # bare filter_field(): force-include this column's full default comparison
+                    comparison_type = self._filter_type(field)
+                    matched_fields.add(field.model_field_name)
+                else:
+                    comparison_type = self._filter_type(field)
                 field.type_ = Optional[comparison_type]  # ty: ignore[invalid-type-form]
+                if declared_filter is not None:
+                    # super() yields GraphQLFieldDefinition (carries graphql_field), though typed as the base.
+                    cast("GraphQLFieldDefinition", field).graphql_field = self._declared_strawberry_field(
+                        declared_filter
+                    )
 
             field.default = UNSET
             field.default_factory = DTOMissing
             yield field
 
+        self._validate_declared_columns(declared_filters, matched_fields, model)
+
+        yield from self._iter_custom_filter_fields(declared_filters, model, dto_config)
+
+        if base is not None and declared_filters:
+            # Declared filter fields supply only a data type, not a final GraphQL type.
+            # Drop their annotations from the base so the strawberry backend does not treat
+            # them as verbatim type overrides; the comparison/custom types injected above win.
+            base.__annotations__ = {
+                annotation_name: annotation_type
+                for annotation_name, annotation_type in inspect.get_annotations(base).items()
+                if annotation_name not in declared_filters
+            }
+
     @override
     def dto_name(
         self, base_name: str, dto_config: DTOConfig, node: Node[Relation[Any, GraphQLFilterDTOT], None] | None = None
diff --git a/src/strawchemy/schema/field.py b/src/strawchemy/schema/field.py
index 0ee25215..2edd5ef4 100644
--- a/src/strawchemy/schema/field.py
+++ b/src/strawchemy/schema/field.py
@@ -1,13 +1,15 @@
 from __future__ import annotations
 
 import dataclasses
+from collections.abc import Callable, Mapping, Sequence
 from functools import cached_property
 from inspect import isclass
-from typing import TYPE_CHECKING, Any, Literal, Optional, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Literal, Optional, TypedDict, TypeVar, cast
 
 from sqlalchemy.orm import InstrumentedAttribute
 from sqlalchemy.sql.elements import UnaryExpression
 from strawberry.annotation import StrawberryAnnotation
+from strawberry.extensions import FieldExtension
 from strawberry.types import get_object_definition
 from strawberry.types.arguments import StrawberryArgument
 from strawberry.types.base import StrawberryOptional
@@ -68,6 +70,40 @@
 T = TypeVar("T", bound="DeclarativeBase")
 
 
+class StrawberryFieldKwargs(TypedDict, total=False):
+    """Input-relevant ``strawberry.field`` args (keys match ``strawberry.field`` for spreading).
+
+    Used by ``Strawchemy.filter_field`` to forward field metadata onto the generated GraphQL
+    input field. Output/resolver-only args are intentionally excluded.
+    """
+
+    name: str | None
+    description: str | None
+    metadata: Mapping[Any, Any] | None
+    deprecation_reason: str | None
+    directives: Sequence[object]
+    graphql_type: Any | None
+
+
+class OutputFieldKwargs(StrawberryFieldKwargs, total=False):
+    """``strawberry.field`` args for output / resolver fields (superset of the input subset).
+
+    Adds the args excluded from input fields: permissions, default value handling, and field
+    extensions. Used by ``Strawchemy.field`` and the mutation methods.
+    """
+
+    permission_classes: list[type[BasePermission]] | None
+    default: Any
+    default_factory: Callable[..., object] | object
+    extensions: list[FieldExtension] | None
+
+
+class MutationFieldKwargs(OutputFieldKwargs, total=False):
+    """Common ``strawberry.field`` + repository args shared by every ``Strawchemy`` mutation method."""
+
+    repository_type: AnyRepositoryType | None
+
+
 class StrawchemyField(StrawberryField):
     """A custom field class for Strawberry GraphQL that allows explicit handling of resolver arguments.
 
diff --git a/src/strawchemy/schema/filters/__init__.py b/src/strawchemy/schema/filters/__init__.py
index 66345e0e..8e078800 100644
--- a/src/strawchemy/schema/filters/__init__.py
+++ b/src/strawchemy/schema/filters/__init__.py
@@ -28,6 +28,7 @@
     TimeDeltaComparison,
     _JSONComparison,
     _SQLiteJSONComparison,
+    _StrawchemyComparison,
     make_full_json_comparison_input,
     make_sqlite_json_comparison_input,
 )
@@ -59,6 +60,7 @@
     "TimeFilter",
     "_JSONComparison",
     "_SQLiteJSONComparison",
+    "_StrawchemyComparison",
     "make_full_json_comparison_input",
     "make_sqlite_json_comparison_input",
 )
diff --git a/src/strawchemy/schema/filters/fields.py b/src/strawchemy/schema/filters/fields.py
new file mode 100644
index 00000000..828ad6da
--- /dev/null
+++ b/src/strawchemy/schema/filters/fields.py
@@ -0,0 +1,61 @@
+"""Marker consumed by the filter factory for fine-grained filters.
+
+`Strawchemy.filter_field()` returns a `FilterFieldMarker`; the field's annotation supplies the
+comparison data type.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any, Literal, Protocol
+
+if TYPE_CHECKING:
+    from sqlalchemy import Dialect, Select
+    from sqlalchemy.orm import DeclarativeBase
+
+    from strawchemy.schema.field import StrawberryFieldKwargs
+
+__all__ = ("VALID_JOINS", "CustomFilterApply", "FilterFieldMarker", "JoinStrategy")
+
+JoinStrategy = Literal["exists", "in"]
+VALID_JOINS: frozenset[str] = frozenset({"exists", "in"})
+
+
+class CustomFilterApply(Protocol):
+    """Callable accepted by ``filter_field(apply=...)``.
+
+    Receives an isolated ``select(model)`` statement and the GraphQL-supplied value, plus the
+    active ``dialect`` and ``model`` as keyword context, and returns the mutated statement. The
+    transpiler passes exactly these keywords; a callback may still declare ``**other_context`` to
+    accept them loosely (a ``**kwargs`` callable structurally satisfies this protocol).
+
+    Note:
+        The callback should only add filtering predicates (``.where(...)``) to the statement. It
+        must not introduce joins or subqueries against the same ``model`` table: the statement is
+        re-aliased and correlated by primary key, so a self-join onto ``model`` could be rewritten
+        ambiguously.
+    """
+
+    def __call__(
+        self,
+        statement: Select[Any],
+        value: Any,
+        /,
+        *,
+        dialect: Dialect,
+        model: type[DeclarativeBase],
+    ) -> Select[Any]: ...
+
+
+@dataclass(frozen=True, slots=True)
+class FilterFieldMarker:
+    """Sentinel describing one declared filter field."""
+
+    ops: tuple[str, ...] | None = None
+    """Selected GraphQL operator names for a restricted field, or ``None``."""
+    apply: CustomFilterApply | None = None
+    """Optional custom filter callable (see :class:`CustomFilterApply`)."""
+    join: JoinStrategy = "exists"
+    """Fold-back strategy for a custom filter (``"exists"`` or ``"in"``)."""
+    field_kwargs: StrawberryFieldKwargs = field(default_factory=dict)
+    """``strawberry.field`` arguments (name, description, metadata, …) for the generated field."""
diff --git a/src/strawchemy/schema/filters/geo.py b/src/strawchemy/schema/filters/geo.py
index 18176519..e9cc8747 100644
--- a/src/strawchemy/schema/filters/geo.py
+++ b/src/strawchemy/schema/filters/geo.py
@@ -11,7 +11,7 @@
 from strawberry import UNSET
 from typing_extensions import override
 
-from strawchemy.schema.filters import FilterProtocol, GraphQLComparison
+from strawchemy.schema.filters import FilterProtocol, GraphQLComparison, _StrawchemyComparison
 from strawchemy.schema.scalars.geo import GeoJSON
 
 __all__ = ("GeoComparison",)
@@ -62,7 +62,7 @@ class GeoComparison(GraphQLComparison):
         within_geometry: Filters for geometries that are within this geometry.
     """
 
-    __strawchemy_filter__ = GeoFilter
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=GeoFilter)
 
     contains_geometry: GeoJSON | None = UNSET
     within_geometry: GeoJSON | None = UNSET
diff --git a/src/strawchemy/schema/filters/inputs.py b/src/strawchemy/schema/filters/inputs.py
index 255aa8c4..d6b1bdea 100644
--- a/src/strawchemy/schema/filters/inputs.py
+++ b/src/strawchemy/schema/filters/inputs.py
@@ -10,13 +10,17 @@
 # ruff: noqa: TC001
 from __future__ import annotations
 
+import inspect
+from dataclasses import dataclass
 from datetime import date, datetime, time, timedelta
 from functools import cache
-from typing import TYPE_CHECKING, Any, Generic, TypeAlias, TypeVar, cast
+from typing import TYPE_CHECKING, Any, ClassVar, Generic, Literal, TypeAlias, TypeVar, cast
 
 import strawberry
 from strawberry import UNSET, Private
+from typing_extensions import Self, assert_never
 
+from strawchemy.exceptions import StrawchemyFieldError
 from strawchemy.schema.filters import (
     ArrayFilter,
     DateFilter,
@@ -32,6 +36,8 @@
 from strawchemy.typing import QueryNodeType
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from sqlalchemy import ColumnElement, Dialect
     from sqlalchemy.orm import QueryableAttribute
 
@@ -52,16 +58,144 @@
 )
 
 T = TypeVar("T")
+_ComparisonT = TypeVar("_ComparisonT", bound="GraphQLComparison")
 GraphQLComparisonT = TypeVar("GraphQLComparisonT", bound="GraphQLComparison")
+
 GraphQLFilter: TypeAlias = "GraphQLComparison | OrderByEnum"
 AnyGraphQLComparison: TypeAlias = "EqualityComparison[Any] | OrderComparison[Any] | TextComparison | DateComparison | TimeComparison | DateTimeComparison | TimeDeltaComparison | ArrayComparison[Any] | _JSONComparison | _SQLiteJSONComparison"
 AnyOrderGraphQLComparison: TypeAlias = (
     "OrderComparison[Any] | TextComparison | DateComparison | TimeComparison | DateTimeComparison | TimeDeltaComparison"
 )
+FilterLabel: TypeAlias = Literal["equality", "order", "string", "list", "date", "time", "interval", "datetime", "json"]
 
 _DESCRIPTION = "Boolean expression to compare {field}. All fields are combined with logical 'AND'"
 
 
+def _scalar(t: Any) -> Any:
+    return t | None  # eq/neq/gt/lt/...  -> T | None  (the common case; Op's default)
+
+
+def _seq(t: Any) -> Any:
+    return list[t] | None  # in/nin/array ops -> list[T] | None
+
+
+@dataclass(frozen=True)
+class Op:
+    """A filter operator: its GraphQL name and how its input value type depends on the data type."""
+
+    graphql_name: str
+    """Name as exposed in the schema (e.g. ``in``)."""
+    value_type: Callable[[Any], Any] = _scalar
+    """``data_type -> annotation`` builder; defaults to scalar ``T | None``."""
+    attr: str | None = None
+    """Python attribute name when it differs from ``graphql_name`` (e.g. ``in_``)."""
+
+    @property
+    def python_attr(self) -> str:
+        return self.attr or self.graphql_name
+
+
+# Equality (order matches EqualityComparison today: eq, neq, is_null, in, nin)
+EQUALITY_OPS = (
+    Op("eq"),
+    Op("neq"),
+    Op("is_null", lambda _t: bool | None),
+    Op("in", _seq, attr="in_"),
+    Op("nin", _seq),
+)
+# Order increment (gt, gte, lt, lte) — all scalar (default value_type)
+ORDER_OPS = (Op("gt"), Op("gte"), Op("lt"), Op("lte"))
+# Text increment (exact order of TextComparison today) — always str
+TEXT_OPS = tuple(
+    Op(n, lambda _t: str | None)
+    for n in (
+        "like",
+        "nlike",
+        "ilike",
+        "nilike",
+        "regexp",
+        "iregexp",
+        "nregexp",
+        "inregexp",
+        "startswith",
+        "endswith",
+        "contains",
+        "istartswith",
+        "iendswith",
+        "icontains",
+    )
+)
+# Array increment (contains, contained_in, overlap) — list-typed
+ARRAY_OPS = (Op("contains", _seq), Op("contained_in", _seq), Op("overlap", _seq))
+# Date parts (int sub-comparisons); referenced lazily because OrderComparison is defined below
+DATE_OPS = tuple(
+    Op(n, lambda _t: OrderComparison[int] | None)
+    for n in ("year", "month", "day", "week_day", "week", "quarter", "iso_year", "iso_week_day")
+)
+# Time parts (int sub-comparisons)
+TIME_OPS = tuple(Op(n, lambda _t: OrderComparison[int] | None) for n in ("hour", "minute", "second"))
+# Interval parts (float sub-comparisons)
+INTERVAL_OPS = tuple(Op(n, lambda _t: OrderComparison[float] | None) for n in ("days", "hours", "minutes", "seconds"))
+
+
+def _filter_description(label: FilterLabel) -> str:
+    """Builds a comparison-input description for the given field label.
+
+    Returns:
+        The full description string, e.g. ``"Boolean expression to compare String fields. ..."``.
+    """
+    match label:
+        case "equality":
+            field = "fields supporting equality comparisons"
+        case "order":
+            field = "fields supporting order comparisons"
+        case "string":
+            field = "String fields"
+        case "list":
+            field = "List fields"
+        case "date":
+            field = "Date fields"
+        case "time":
+            field = "Time fields"
+        case "interval":
+            field = "Interval fields"
+        case "datetime":
+            field = "DateTime fields"
+        case "json":
+            field = "JSON fields"
+        case _:
+            assert_never(label)
+    return _DESCRIPTION.format(field=field)
+
+
+def comparison_input(
+    name: str, *, data_type: Any = T, description: str = ""
+) -> Callable[[type[_ComparisonT]], type[_ComparisonT]]:
+    """Generates a comparison class's own operator fields from its ``__strawchemy_comparison__`` and decorates it.
+
+    ``data_type`` is the TypeVar ``T`` for generic comparisons, or a concrete type (e.g. ``str``).
+    """
+
+    def deco(cls: type[_ComparisonT]) -> type[_ComparisonT]:
+        annotations, attributes = cls._field_namespace(data_type, cls.__strawchemy_comparison__.operators)
+        cls.__annotations__ = {**inspect.get_annotations(cls), **annotations}
+        for attr, value in attributes.items():
+            setattr(cls, attr, value)
+        return strawberry.input(name=name, description=description)(cls)
+
+    return deco
+
+
+@dataclass(frozen=True, slots=True)
+class _StrawchemyComparison:
+    """Static metadata for a comparison class."""
+
+    filter: type[FilterProtocol]
+    """The query-time filter that turns this comparison into SQL expressions."""
+    operators: tuple[Op, ...] = ()
+    """The comparison's own operators (merged across the MRO by ``_all_operators``)."""
+
+
 class GraphQLComparison:
     """Base class for GraphQL comparison filters.
 
@@ -76,12 +210,86 @@ class GraphQLComparison:
     """
 
     __strawchemy_field_node__: Private[QueryNodeType | None] = None
-    __strawchemy_filter__: Private[type[FilterProtocol]]
+    __strawchemy_comparison__: ClassVar[_StrawchemyComparison]
+    _restricted_cache: ClassVar[dict[tuple[type[Any], Any, tuple[str, ...]], type[GraphQLComparison]]] = {}
+
+    @classmethod
+    def _all_operators(cls) -> tuple[Op, ...]:
+        """Full operator set for this comparison, collected base-first across the MRO (deduped)."""
+        seen: set[str] = set()
+        collected: list[Op] = []
+        for klass in reversed(inspect.getmro(cls)):
+            definition: _StrawchemyComparison | None = klass.__dict__.get("__strawchemy_comparison__")
+            for op in definition.operators if definition is not None else ():
+                if op.graphql_name not in seen:
+                    seen.add(op.graphql_name)
+                    collected.append(op)
+        return tuple(collected)
+
+    @classmethod
+    def restricted(cls, data_type: Any, ops: tuple[str, ...]) -> type[Self]:
+        """Builds (and caches) a comparison input over ``data_type`` exposing only ``ops``.
+
+        The result is a standalone strawberry input carrying every operator attribute (unselected
+        ones defaulted ``UNSET``) so the query-time ``*Filter.to_expressions`` keep working, and the
+        same ``__strawchemy_comparison__.filter`` as ``cls``.
+
+        Args:
+            data_type: The scalar type the operators compare against.
+            ops: Selected GraphQL operator names (order-independent).
+
+        Returns:
+            A strawberry input type restricted to the selected operators.
+
+        Raises:
+            StrawchemyFieldError: If an operator is not defined for this comparison.
+        """
+        chosen_ops = tuple(sorted(set(ops)))
+        cache_key = (cls, data_type, chosen_ops)
+        if (cached := GraphQLComparison._restricted_cache.get(cache_key)) is not None:
+            return cast("type[Self]", cached)
+
+        all_ops = cls._all_operators()
+        ops_by_name = {op.graphql_name: op for op in all_ops}
+
+        if invalid_ops := [op for op in chosen_ops if op not in ops_by_name]:
+            msg = f"Operator(s) {invalid_ops} not valid for {cls.__name__}; allowed: {sorted(ops_by_name)}"
+            raise StrawchemyFieldError(msg)
+
+        annotations, attributes = cls._field_namespace(data_type, tuple(ops_by_name[op] for op in chosen_ops))
+        namespace: dict[str, Any] = {
+            "__annotations__": annotations,
+            "__strawchemy_comparison__": _StrawchemyComparison(filter=cls.__strawchemy_comparison__.filter),
+            **attributes,
+        }
+        for op in all_ops:  # unselected ops present as UNSET so query-time filters work
+            namespace.setdefault(op.python_attr, UNSET)
+
+        # Build restricted type
+        suffix = "".join(part.capitalize() for op in chosen_ops for part in op.split("_"))
+        type_name = f"{cls.__name__}{data_type.__name__.capitalize()}{suffix}"
+        built = cast(
+            "type[Self]",
+            strawberry.input(name=type_name, description=f"{cls.__name__} restricted to {', '.join(chosen_ops)}")(
+                type(type_name, (GraphQLComparison,), namespace)
+            ),
+        )
+        GraphQLComparison._restricted_cache[cache_key] = built
+        return built
+
+    @classmethod
+    def _field_namespace(cls, data_type: Any, ops: tuple[Op, ...]) -> tuple[dict[str, Any], dict[str, Any]]:
+        """Builds (annotations, attributes) for ``ops`` resolved against ``data_type``."""
+        annotations = {op.python_attr: op.value_type(data_type) for op in ops}
+        attributes = {
+            op.python_attr: (strawberry.field(name=op.graphql_name, default=UNSET) if op.attr else UNSET) for op in ops
+        }
+        return annotations, attributes
 
     def to_expressions(
         self, dialect: Dialect, model_attribute: QueryableAttribute[Any] | ColumnElement[Any]
     ) -> list[ColumnElement[bool]]:
-        return self.__strawchemy_filter__(self).to_expressions(dialect, model_attribute)
+        return self.__strawchemy_comparison__.filter(self).to_expressions(dialect, model_attribute)
 
     @property
     def field_node(self) -> QueryNodeType:
@@ -94,192 +302,113 @@ def field_node(self, value: QueryNodeType) -> None:
         self.__strawchemy_field_node__ = value
 
 
-@strawberry.input(
-    name="GenericComparison", description=_DESCRIPTION.format(field="fields supporting equality comparisons")
-)
+@comparison_input("GenericComparison", description=_filter_description("equality"))
 class EqualityComparison(GraphQLComparison, Generic[T]):
-    """Generic comparison class for GraphQL filters.
-
-    This class provides a set of generic comparison operators that can be
-    used to filter data based on equality, inequality, null checks, and
-    inclusion in a list.
-
-    Attributes:
-        eq: Filters for values equal to this.
-        neq: Filters for values not equal to this.
-        is_null: Filters for null values if True, or non-null values if False.
-        in: Filters for values present in this list.
-        nin: Filters for values not present in this list.
-    """
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=EqualityFilter, operators=EQUALITY_OPS)
 
-    __strawchemy_filter__ = EqualityFilter
+    if TYPE_CHECKING:
+        # Generated at runtime from ``__strawchemy_comparison__.operators``; declared for type checkers.
+        eq: T | None
+        neq: T | None
+        is_null: bool | None
+        in_: list[T] | None
+        nin: list[T] | None
 
-    eq: T | None = UNSET
-    neq: T | None = UNSET
-    is_null: bool | None = UNSET
-    in_: list[T] | None = strawberry.field(name="in", default=UNSET)
-    nin: list[T] | None = UNSET
 
-
-@strawberry.input(name="OrderComparison", description=_DESCRIPTION.format(field="fields supporting order comparisons"))
+@comparison_input("OrderComparison", description=_filter_description("order"))
 class OrderComparison(EqualityComparison[T]):
-    """Order comparison class for GraphQL filters.
-
-    This class provides a set of numeric comparison operators that can be
-    used to filter data based on greater than, less than, and equality.
-
-    Attributes:
-        gt: Filters for values greater than this.
-        gte: Filters for values greater than or equal to this.
-        lt: Filters for values less than this.
-        lte: Filters for values less than or equal to this.
-    """
-
-    __strawchemy_filter__ = OrderFilter
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=OrderFilter, operators=ORDER_OPS)
 
-    gt: T | None = UNSET
-    gte: T | None = UNSET
-    lt: T | None = UNSET
-    lte: T | None = UNSET
+    if TYPE_CHECKING:
+        gt: T | None
+        gte: T | None
+        lt: T | None
+        lte: T | None
 
 
-@strawberry.input(name="TextComparison", description=_DESCRIPTION.format(field="String fields"))
+@comparison_input("TextComparison", data_type=str, description=_filter_description("string"))
 class TextComparison(OrderComparison[str]):
-    """Text comparison class for GraphQL filters.
-
-    This class provides a set of text comparison operators that can be
-    used to filter data based on various string matching patterns.
-
-    Attributes:
-        like: Filters for values that match this SQL LIKE pattern.
-        nlike: Filters for values that do not match this SQL LIKE pattern.
-        ilike: Filters for values that match this case-insensitive SQL LIKE pattern.
-        nilike: Filters for values that do not match this case-insensitive SQL LIKE pattern.
-        regexp: Filters for values that match this regular expression.
-        nregexp: Filters for values that do not match this regular expression.
-        startswith: Filters for values that start with this string.
-        endswith: Filters for values that end with this string.
-        contains: Filters for values that contain this string.
-        istartswith: Filters for values that start with this string (case-insensitive).
-        iendswith: Filters for values that end with this string (case-insensitive).
-        icontains: Filters for values that contain this string (case-insensitive).
-    """
-
-    __strawchemy_filter__ = TextFilter
-
-    like: str | None = UNSET
-    nlike: str | None = UNSET
-    ilike: str | None = UNSET
-    nilike: str | None = UNSET
-    regexp: str | None = UNSET
-    iregexp: str | None = UNSET
-    nregexp: str | None = UNSET
-    inregexp: str | None = UNSET
-    startswith: str | None = UNSET
-    endswith: str | None = UNSET
-    contains: str | None = UNSET
-    istartswith: str | None = UNSET
-    iendswith: str | None = UNSET
-    icontains: str | None = UNSET
-
-
-@strawberry.input(name="ArrayComparison", description=_DESCRIPTION.format(field="List fields"))
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=TextFilter, operators=TEXT_OPS)
+
+    if TYPE_CHECKING:
+        like: str | None
+        nlike: str | None
+        ilike: str | None
+        nilike: str | None
+        regexp: str | None
+        iregexp: str | None
+        nregexp: str | None
+        inregexp: str | None
+        startswith: str | None
+        endswith: str | None
+        contains: str | None
+        istartswith: str | None
+        iendswith: str | None
+        icontains: str | None
+
+
+@comparison_input("ArrayComparison", description=_filter_description("list"))
 class ArrayComparison(EqualityComparison[T], Generic[T]):
-    """Postgres array comparison class for GraphQL filters.
-
-    This class provides a set of array comparison operators that can be
-    used to filter data based on containment, overlap, and other
-    array-specific properties.
-
-    Attributes:
-        contains: Filters for array values that contain all elements in this list.
-        contained_in: Filters for array values that are contained in this list.
-        overlap: Filters for array values that have any elements in common with this list.
-    """
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=ArrayFilter, operators=ARRAY_OPS)
 
-    __strawchemy_filter__ = ArrayFilter
+    if TYPE_CHECKING:
+        contains: list[T] | None
+        contained_in: list[T] | None
+        overlap: list[T] | None
 
-    contains: list[T] | None = UNSET
-    contained_in: list[T] | None = UNSET
-    overlap: list[T] | None = UNSET
 
-
-@strawberry.input(name="DateComparison", description=_DESCRIPTION.format(field="Date fields"))
+@comparison_input("DateComparison", data_type=date, description=_filter_description("date"))
 class DateComparison(OrderComparison[date]):
-    """Date comparison class for GraphQL filters.
-
-    This class provides a set of date component comparison operators that
-    can be used to filter data based on specific parts of a date.
-
-    Attributes:
-        year: Filters based on the year.
-        month: Filters based on the month.
-        day: Filters based on the day.
-        week_day: Filters based on the day of the week.
-        week: Filters based on the week number.
-        quarter: Filters based on the quarter of the year.
-        iso_year: Filters based on the ISO year.
-        iso_week_day: Filters based on the ISO day of the week.
-    """
-
-    __strawchemy_filter__ = DateFilter
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=DateFilter, operators=DATE_OPS)
 
-    year: OrderComparison[int] | None = UNSET
-    month: OrderComparison[int] | None = UNSET
-    day: OrderComparison[int] | None = UNSET
-    week_day: OrderComparison[int] | None = UNSET
-    week: OrderComparison[int] | None = UNSET
-    quarter: OrderComparison[int] | None = UNSET
-    iso_year: OrderComparison[int] | None = UNSET
-    iso_week_day: OrderComparison[int] | None = UNSET
+    if TYPE_CHECKING:
+        year: OrderComparison[int] | None
+        month: OrderComparison[int] | None
+        day: OrderComparison[int] | None
+        week_day: OrderComparison[int] | None
+        week: OrderComparison[int] | None
+        quarter: OrderComparison[int] | None
+        iso_year: OrderComparison[int] | None
+        iso_week_day: OrderComparison[int] | None
 
 
-@strawberry.input(name="TimeComparison", description=_DESCRIPTION.format(field="Time fields"))
+@comparison_input("TimeComparison", data_type=time, description=_filter_description("time"))
 class TimeComparison(OrderComparison[time]):
-    """Time comparison class for GraphQL filters.
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=TimeFilter, operators=TIME_OPS)
 
-    This class provides a set of time component comparison operators that
-    can be used to filter data based on specific parts of a time.
+    if TYPE_CHECKING:
+        hour: OrderComparison[int] | None
+        minute: OrderComparison[int] | None
+        second: OrderComparison[int] | None
 
-    Attributes:
-        hour: Filters based on the hour.
-        minute: Filters based on the minute.
-        second: Filters based on the second.
-    """
-
-    __strawchemy_filter__ = TimeFilter
-
-    hour: OrderComparison[int] | None = UNSET
-    minute: OrderComparison[int] | None = UNSET
-    second: OrderComparison[int] | None = UNSET
 
-
-@strawberry.input(name="IntervalComparison", description=_DESCRIPTION.format(field="Interval fields"))
+@comparison_input("IntervalComparison", data_type=timedelta, description=_filter_description("interval"))
 class TimeDeltaComparison(OrderComparison[timedelta]):
-    __strawchemy_filter__ = TimeDeltaFilter
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=TimeDeltaFilter, operators=INTERVAL_OPS)
 
-    days: OrderComparison[float] | None = UNSET
-    hours: OrderComparison[float] | None = UNSET
-    minutes: OrderComparison[float] | None = UNSET
-    seconds: OrderComparison[float] | None = UNSET
+    if TYPE_CHECKING:
+        days: OrderComparison[float] | None
+        hours: OrderComparison[float] | None
+        minutes: OrderComparison[float] | None
+        seconds: OrderComparison[float] | None
 
 
-@strawberry.input(name="DateTimeComparison", description=_DESCRIPTION.format(field="DateTime fields"))
+@comparison_input("DateTimeComparison", data_type=datetime, description=_filter_description("datetime"))
 class DateTimeComparison(OrderComparison[datetime]):
-    __strawchemy_filter__ = DateTimeFilter
-
-    year: OrderComparison[int] | None = UNSET
-    month: OrderComparison[int] | None = UNSET
-    day: OrderComparison[int] | None = UNSET
-    week_day: OrderComparison[int] | None = UNSET
-    week: OrderComparison[int] | None = UNSET
-    quarter: OrderComparison[int] | None = UNSET
-    iso_year: OrderComparison[int] | None = UNSET
-    iso_week_day: OrderComparison[int] | None = UNSET
-
-    hour: OrderComparison[int] | None = UNSET
-    minute: OrderComparison[int] | None = UNSET
-    second: OrderComparison[int] | None = UNSET
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=DateTimeFilter, operators=(*DATE_OPS, *TIME_OPS))
+
+    if TYPE_CHECKING:
+        year: OrderComparison[int] | None
+        month: OrderComparison[int] | None
+        day: OrderComparison[int] | None
+        week_day: OrderComparison[int] | None
+        week: OrderComparison[int] | None
+        quarter: OrderComparison[int] | None
+        iso_year: OrderComparison[int] | None
+        iso_week_day: OrderComparison[int] | None
+        hour: OrderComparison[int] | None
+        minute: OrderComparison[int] | None
+        second: OrderComparison[int] | None
 
 
 class _JSONComparison(EqualityComparison[dict[str, Any]]):
@@ -297,7 +426,7 @@ class _JSONComparison(EqualityComparison[dict[str, Any]]):
         has_key_any: Filters for JSON values that have any of these keys.
     """
 
-    __strawchemy_filter__ = JSONFilter
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=JSONFilter)
 
     contains: dict[str, Any] | None = UNSET
     contained_in: dict[str, Any] | None = UNSET
@@ -321,7 +450,7 @@ class _SQLiteJSONComparison(EqualityComparison[dict[str, Any]]):
         has_key_any: Filters for JSON values that have any of these keys.
     """
 
-    __strawchemy_filter__ = JSONFilter
+    __strawchemy_comparison__ = _StrawchemyComparison(filter=JSONFilter)
 
     has_key: str | None = UNSET
     has_key_all: list[str] | None = UNSET
@@ -332,7 +461,7 @@ class _SQLiteJSONComparison(EqualityComparison[dict[str, Any]]):
 def make_full_json_comparison_input() -> type[_JSONComparison]:
     return cast(
         "type[_JSONComparison]",
-        strawberry.input(name="JSONComparison", description=_DESCRIPTION.format(field="JSON fields"))(_JSONComparison),
+        strawberry.input(name="JSONComparison", description=_filter_description("json"))(_JSONComparison),
     )
 
 
@@ -340,7 +469,5 @@ def make_full_json_comparison_input() -> type[_JSONComparison]:
 def make_sqlite_json_comparison_input() -> type[_SQLiteJSONComparison]:
     return cast(
         "type[_SQLiteJSONComparison]",
-        strawberry.input(name="JSONComparison", description=_DESCRIPTION.format(field="JSON fields"))(
-            _SQLiteJSONComparison
-        ),
+        strawberry.input(name="JSONComparison", description=_filter_description("json"))(_SQLiteJSONComparison),
     )
diff --git a/src/strawchemy/schema/mutation/field_builder.py b/src/strawchemy/schema/mutation/field_builder.py
index 699287d4..9d9ded52 100644
--- a/src/strawchemy/schema/mutation/field_builder.py
+++ b/src/strawchemy/schema/mutation/field_builder.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import dataclasses
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
@@ -16,16 +15,12 @@
 )
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Mapping, Sequence
-
-    from strawberry import BasePermission
-    from strawberry.extensions.field_extension import FieldExtension
+    from collections.abc import Callable
 
     from strawchemy.config.base import StrawchemyConfig
     from strawchemy.schema.factories import DistinctOnEnumFactory
     from strawchemy.schema.factories.inputs import BooleanFilterFactory, OrderByFactory
     from strawchemy.schema.mutation.input import EventRegistry
-    from strawchemy.typing import AnyRepositoryType
 
 
 @dataclass
@@ -55,18 +50,8 @@ def build(
         ],
         resolver: Any | None = None,
         *,
-        repository_type: AnyRepositoryType | None = None,
         graphql_type: Any | None = None,
-        name: str | None = None,
-        description: str | None = None,
-        permission_classes: list[type[BasePermission]] | None = None,
-        deprecation_reason: str | None = None,
-        default: Any = dataclasses.MISSING,
-        default_factory: Callable[..., object] | object = dataclasses.MISSING,
-        metadata: Mapping[Any, Any] | None = None,
-        directives: Sequence[object] = (),
-        extensions: list[FieldExtension] | None = None,
-        **field_specific_kwargs: Any,
+        **field_kwargs: Any,
     ) -> Any:
         """Build a mutation field with common configuration.
 
@@ -74,20 +59,11 @@ def build(
             field_class: The specific mutation field class to instantiate
                 (e.g., StrawchemyCreateMutationField).
             resolver: An optional custom resolver function for the mutation.
-            repository_type: An optional custom strawberry class. Defaults to
-                the strawberry configured in StrawchemyConfig.
             graphql_type: The GraphQL return type of the mutation.
-            name: The name of the GraphQL mutation field.
-            description: The description of the GraphQL mutation field.
-            permission_classes: A list of permission classes for the field.
-            deprecation_reason: The reason for deprecating the field.
-            default: The default value for the field.
-            default_factory: A factory function to generate the default value.
-            metadata: Additional metadata for the field.
-            directives: A sequence of directives for the field.
-            extensions: A list of Strawberry FieldExtensions.
-            **field_specific_kwargs: Additional keyword arguments specific to
-                the field type (e.g., input_type, filter_input, update_fields, etc.).
+            **field_kwargs: The common ``strawberry.field`` / repository arguments
+                (see ``MutationFieldKwargs``) merged with any field-specific arguments
+                (e.g., input_type, filter_input, update_fields). ``name`` maps to the
+                GraphQL field name.
 
         Returns:
             A configured mutation field instance, either wrapped with the resolver
@@ -95,33 +71,25 @@ def build(
         """
         namespace = self.registry_namespace_getter()
         type_annotation = StrawberryAnnotation.from_annotation(graphql_type, namespace) if graphql_type else None
+        graphql_name = field_kwargs.pop("name", None)
 
         # Inject the shared registry only for input mutation fields (create/update/upsert),
         # not for the delete field which shares the `input_type` kwarg name for filter types.
         if issubclass(
             field_class, (StrawchemyCreateMutationField, StrawchemyUpdateMutationField, StrawchemyUpsertMutationField)
         ):
-            field_specific_kwargs.setdefault("event_registry", self.event_registry)
+            field_kwargs.setdefault("event_registry", self.event_registry)
 
         field = field_class(
             config=self.config,
-            repository_type=repository_type,
             python_name=None,
-            graphql_name=name,
+            graphql_name=graphql_name,
             type_annotation=type_annotation,
             is_subscription=False,
-            permission_classes=permission_classes or [],
-            deprecation_reason=deprecation_reason,
-            default=default,
-            default_factory=default_factory,
-            metadata=metadata,
-            directives=directives,
-            extensions=extensions or [],
             registry_namespace=namespace,
-            description=description,
             order_by_factory=self.order_by_factory,
             filter_factory=self.filter_factory,
             distinct_on_factory=self.distinct_on_factory,
-            **field_specific_kwargs,
+            **field_kwargs,
         )
         return field(resolver) if resolver else field
diff --git a/src/strawchemy/transpiler/_planner.py b/src/strawchemy/transpiler/_planner.py
index 86b2e9b8..d1ade650 100644
--- a/src/strawchemy/transpiler/_planner.py
+++ b/src/strawchemy/transpiler/_planner.py
@@ -11,7 +11,7 @@
 from dataclasses import dataclass, field, replace
 from typing import TYPE_CHECKING, Any, Generic, cast
 
-from sqlalchemy import and_, func, inspect, not_, null, or_, select, true
+from sqlalchemy import and_, exists, func, inspect, not_, null, or_, select, true, tuple_
 from sqlalchemy.orm import Mapper, RelationshipProperty, aliased, class_mapper, contains_eager, load_only, raiseload
 from sqlalchemy.sql.util import ClauseAdapter
 
@@ -20,6 +20,7 @@
 from strawchemy.dto.inspectors.sqlalchemy import SQLAlchemyInspector
 from strawchemy.dto.strawberry import (
     AggregationFilter,
+    CustomFilter,
     Filter,
     OrderByEnum,
     OrderByRelationFilterDTO,
@@ -418,6 +419,53 @@ def _to_expressions(
             expressions.append(attribute.is_not(null()))
         return expressions
 
+    @staticmethod
+    def _custom_filter_expression(custom: CustomFilter, context: PlanContext[Any]) -> ColumnElement[bool]:
+        """Folds a custom-apply filter into a correlated EXISTS/IN predicate.
+
+        Builds an isolated ``select(model)``, lets the callback mutate it, then correlates it to
+        the outer query on primary-key equality so the result is a single boolean predicate that
+        composes under AND/OR/NOT.
+
+        Args:
+            custom: The set custom filter value (callable, value, strategy, model node).
+            context: The shared planning context (``aliases``, ``dialect`` read here).
+
+        Returns:
+            A boolean SQLAlchemy expression.
+        """
+        model = custom.field_node.value.model
+        mapper = class_mapper(model)
+        inner_pks = SQLAlchemyInspector.pk_attributes(mapper)
+        outer_pks = context.aliases.aliased_id_attributes(custom.field_node)
+
+        isolated = custom.apply(select(model), custom.value, dialect=context.dialect, model=model)
+
+        if custom.join == "in":
+            inner_select = isolated.with_only_columns(*inner_pks)
+            if len(outer_pks) == 1:
+                # Scalar IN is more portable/optimizable than a single-element tuple IN.
+                return outer_pks[0].in_(inner_select)
+            return tuple_(*outer_pks).in_(inner_select)
+
+        outer_alias = (
+            context.aliases.root_alias
+            if custom.field_node.is_root
+            else context.aliases.alias_from_relation_node(custom.field_node, "target")
+        )
+
+        # The outer query aliases the root model to a name equal to its table name, so the inner
+        # subquery must use a distinct alias; otherwise the PK-equality correlation collapses to a
+        # tautology against the same table and the EXISTS matches every outer row. An unnamed alias
+        # lets SQLAlchemy generate a guaranteed-unique name (also safe for repeated/nested filters).
+        inner_alias = aliased(mapper, flat=True)
+        adapter = ClauseAdapter(inspect(inner_alias).selectable)
+        adapted_pks = [adapter.traverse(pk.__clause_element__()) for pk in inner_pks]
+        adapted_select = adapter.traverse(isolated.with_only_columns(*adapted_pks))
+
+        correlation = and_(*[inner == outer for inner, outer in zip(adapted_pks, outer_pks, strict=True)])
+        return exists(adapted_select.where(correlation)).correlate(outer_alias)
+
     @staticmethod
     def _aggregation_filter(
         aggregation: AggregationFilter,
@@ -457,7 +505,7 @@ def _aggregation_filter(
 
     @staticmethod
     def _gather_conjunctions(
-        query: Sequence[Filter | AggregationFilter | GraphQLComparison],
+        query: Sequence[Filter | AggregationFilter | GraphQLComparison | CustomFilter],
         context: PlanContext[Any],
         agg_plan: AggregationPlan,
         emitted_agg_joins: set[QueryNodeType],
@@ -495,6 +543,9 @@ def _gather_conjunctions(
             elif isinstance(value, GraphQLComparison):
                 node_path = value.field_node.path_from_root()
                 bool_expressions.extend(FilterPlan._to_expressions(context, value, not_null_check=not_null_check))
+            elif isinstance(value, CustomFilter):
+                node_path = value.field_node.path_from_root()
+                bool_expressions.append(FilterPlan._custom_filter_expression(value, context))
             else:
                 conjunction = FilterPlan._conjunctions(
                     value, context, agg_plan, emitted_agg_joins, where_function_nodes, not_null_check
diff --git a/src/strawchemy/utils/annotation.py b/src/strawchemy/utils/annotation.py
index 9393b17f..60040286 100644
--- a/src/strawchemy/utils/annotation.py
+++ b/src/strawchemy/utils/annotation.py
@@ -55,6 +55,14 @@ def get_annotations(obj: Any) -> dict[str, Any]:
     return inspect.get_annotations(obj)
 
 
+def get_origin_or_self(annotation: Any) -> Any:
+    """Returns the unsubscripted origin of a parametrized type, or the object itself.
+
+    ``OrderComparison[int]`` -> ``OrderComparison``; ``TextComparison`` -> ``TextComparison``.
+    """
+    return get_origin(annotation) or annotation
+
+
 def inner_types(annotation: Any) -> tuple[Any, ...]:
     """Flatten a generic annotation to its innermost leaf types.
 
diff --git a/tests/integration/__snapshots__/test_fine_grained_filters.ambr b/tests/integration/__snapshots__/test_fine_grained_filters.ambr
new file mode 100644
index 00000000..7da0dc59
--- /dev/null
+++ b/tests/integration/__snapshots__/test_fine_grained_filters.ambr
@@ -0,0 +1,151 @@
+# serializer version: 1
+# name: test_custom_apply_filter_in_strategy[session-tracked-async-aiosqlite_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE fruit.id IN (
+          SELECT fruit.id
+            FROM fruit
+           WHERE fruit.sweetness >= ?
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_in_strategy[session-tracked-async-asyncmy_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE fruit.id IN (
+          SELECT fruit.id
+            FROM fruit
+           WHERE fruit.sweetness >= %s
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_in_strategy[session-tracked-async-asyncpg_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE fruit.id IN (
+          SELECT fruit.id
+            FROM fruit
+           WHERE fruit.sweetness >= $1::INTEGER
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_in_strategy[session-tracked-async-psycopg_async_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE fruit.id IN (
+          SELECT fruit.id
+            FROM fruit
+           WHERE fruit.sweetness >= %(sweetness_1)s::INTEGER
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_in_strategy[session-tracked-sync-psycopg_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE fruit.id IN (
+          SELECT fruit.id
+            FROM fruit
+           WHERE fruit.sweetness >= %(sweetness_1)s::INTEGER
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_in_strategy[session-tracked-sync-sqlite_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE fruit.id IN (
+          SELECT fruit.id
+            FROM fruit
+           WHERE fruit.sweetness >= ?
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_sweeter_than[session-tracked-async-aiosqlite_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE EXISTS (
+          SELECT fruit_1.id
+            FROM fruit AS fruit_1
+           WHERE fruit_1.sweetness >= ?
+             AND fruit_1.id = fruit.id
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_sweeter_than[session-tracked-async-asyncmy_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE EXISTS (
+          SELECT fruit_1.id
+            FROM fruit AS fruit_1
+           WHERE fruit_1.sweetness >= %s
+             AND fruit_1.id = fruit.id
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_sweeter_than[session-tracked-async-asyncpg_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE EXISTS (
+          SELECT fruit_1.id
+            FROM fruit AS fruit_1
+           WHERE fruit_1.sweetness >= $1::INTEGER
+             AND fruit_1.id = fruit.id
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_sweeter_than[session-tracked-async-psycopg_async_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE EXISTS (
+          SELECT fruit_1.id
+            FROM fruit AS fruit_1
+           WHERE fruit_1.sweetness >= %(sweetness_1)s::INTEGER
+             AND fruit_1.id = fruit.id
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_sweeter_than[session-tracked-sync-psycopg_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE EXISTS (
+          SELECT fruit_1.id
+            FROM fruit AS fruit_1
+           WHERE fruit_1.sweetness >= %(sweetness_1)s::INTEGER
+             AND fruit_1.id = fruit.id
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
+# name: test_custom_apply_filter_sweeter_than[session-tracked-sync-sqlite_engine]
+  '''
+  SELECT fruit.id
+    FROM fruit AS fruit
+   WHERE EXISTS (
+          SELECT fruit_1.id
+            FROM fruit AS fruit_1
+           WHERE fruit_1.sweetness >= ?
+             AND fruit_1.id = fruit.id
+         )
+   ORDER BY fruit.id ASC
+  '''
+# ---
diff --git a/tests/integration/test_fine_grained_filters.py b/tests/integration/test_fine_grained_filters.py
new file mode 100644
index 00000000..1ae95637
--- /dev/null
+++ b/tests/integration/test_fine_grained_filters.py
@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from tests.utils import maybe_async
+
+if TYPE_CHECKING:
+    from syrupy.assertion import SnapshotAssertion
+
+    from tests.integration.fixtures import QueryTracker
+    from tests.integration.typing import RawRecordData
+    from tests.typing import AnyQueryExecutor
+
+
+@pytest.mark.snapshot
+async def test_custom_apply_filter_sweeter_than(
+    any_query: AnyQueryExecutor, raw_fruits: RawRecordData, query_tracker: QueryTracker, sql_snapshot: SnapshotAssertion
+) -> None:
+    threshold = 5
+    expected_ids = {fruit["id"] for fruit in raw_fruits if fruit["sweetness"] >= threshold}
+    query = f"""
+        {{
+            fruitsFineGrained(filter: {{ sweeterThan: {threshold} }}) {{
+                id
+            }}
+        }}
+    """
+    result = await maybe_async(any_query(query))
+    assert not result.errors
+    assert result.data is not None
+    assert {row["id"] for row in result.data["fruitsFineGrained"]} == expected_ids
+    assert query_tracker[0].statement_formatted == sql_snapshot
+
+
+@pytest.mark.snapshot
+async def test_custom_apply_filter_in_strategy(
+    any_query: AnyQueryExecutor, raw_fruits: RawRecordData, query_tracker: QueryTracker, sql_snapshot: SnapshotAssertion
+) -> None:
+    threshold = 5
+    expected_ids = {fruit["id"] for fruit in raw_fruits if fruit["sweetness"] >= threshold}
+    query = f"""
+        {{
+            fruitsFineGrained(filter: {{ sweeterThanIn: {threshold} }}) {{
+                id
+            }}
+        }}
+    """
+    result = await maybe_async(any_query(query))
+    assert not result.errors
+    assert result.data is not None
+    assert {row["id"] for row in result.data["fruitsFineGrained"]} == expected_ids
+    assert query_tracker[0].statement_formatted == sql_snapshot
+
+
+async def test_restricted_operator_only_selected_exposed(any_query: AnyQueryExecutor) -> None:
+    # `contains` is not in the selected ops {eq, like} -> GraphQL validation error.
+    query = """
+        {
+            fruitsFineGrained(filter: { name: { contains: "x" } }) {
+                id
+            }
+        }
+    """
+    result = await maybe_async(any_query(query))
+    assert result.errors
+
+
+async def test_custom_apply_filter_under_or(any_query: AnyQueryExecutor, raw_fruits: RawRecordData) -> None:
+    threshold = 8
+    target_name = raw_fruits[0]["name"]
+    expected_ids = {
+        fruit["id"] for fruit in raw_fruits if fruit["sweetness"] >= threshold or fruit["name"] == target_name
+    }
+    query = f"""
+        {{
+            fruitsFineGrained(
+                filter: {{
+                    _or: [
+                        {{ sweeterThan: {threshold} }}
+                        {{ name: {{ eq: "{target_name}" }} }}
+                    ]
+                }}
+            ) {{
+                id
+            }}
+        }}
+    """
+    result = await maybe_async(any_query(query))
+    assert not result.errors
+    assert result.data is not None
+    assert {row["id"] for row in result.data["fruitsFineGrained"]} == expected_ids
diff --git a/tests/integration/types/mysql.py b/tests/integration/types/mysql.py
index 41b56571..95cc75b4 100644
--- a/tests/integration/types/mysql.py
+++ b/tests/integration/types/mysql.py
@@ -18,6 +18,7 @@
     StrawchemyAsyncRepository,
     StrawchemyConfig,
     StrawchemySyncRepository,
+    TextComparison,
     ValidationErrorType,
 )
 from strawchemy.schema.pagination import DefaultOffsetPagination
@@ -159,6 +160,17 @@ class FruitTypeWithPaginationAndOrderBy: ...
 class FruitFilter: ...
 
 
+def _fruit_sweeter_than(statement: Select[tuple[Fruit]], value: int, **_ctx: Any) -> Select[tuple[Fruit]]:
+    return statement.where(Fruit.sweetness >= value)
+
+
+@strawchemy.filter(Fruit, include=["id", "name", "sweetness"], name="FruitFineGrainedFilter")
+class FruitFineGrainedFilter:
+    name: TextComparison = strawchemy.filter_field(ops=["eq", "like"])
+    sweeter_than: int = strawchemy.filter_field(apply=_fruit_sweeter_than)
+    sweeter_than_in: int = strawchemy.filter_field(apply=_fruit_sweeter_than, join="in")
+
+
 @strawchemy.order(Fruit, include="all", scope="schema")
 class FruitOrderBy: ...
 
@@ -303,6 +315,9 @@ class AsyncQuery:
     fruits: list[FruitType] = strawchemy.field(
         filter_input=FruitFilter, order_by_input=FruitOrderBy, repository_type=StrawchemyAsyncRepository
     )
+    fruits_fine_grained: list[FruitType] = strawchemy.field(
+        filter_input=FruitFineGrainedFilter, repository_type=StrawchemyAsyncRepository
+    )
     fruits_paginated: list[FruitTypeWithPaginationAndOrderBy] = strawchemy.field(
         filter_input=FruitFilter,
         pagination=True,
@@ -427,6 +442,9 @@ class SyncQuery:
     fruits: list[FruitType] = strawchemy.field(
         filter_input=FruitFilter, order_by_input=FruitOrderBy, repository_type=StrawchemySyncRepository
     )
+    fruits_fine_grained: list[FruitType] = strawchemy.field(
+        filter_input=FruitFineGrainedFilter, repository_type=StrawchemySyncRepository
+    )
     fruits_paginated: list[FruitTypeWithPaginationAndOrderBy] = strawchemy.field(
         filter_input=FruitFilter,
         pagination=True,
diff --git a/tests/integration/types/postgres.py b/tests/integration/types/postgres.py
index 112b78ce..5992c66d 100644
--- a/tests/integration/types/postgres.py
+++ b/tests/integration/types/postgres.py
@@ -17,6 +17,7 @@
     Strawchemy,
     StrawchemyAsyncRepository,
     StrawchemySyncRepository,
+    TextComparison,
     ValidationErrorType,
 )
 from strawchemy.schema.pagination import DefaultOffsetPagination
@@ -167,6 +168,17 @@ class FruitTypeWithPaginationAndOrderBy: ...
 class FruitFilter: ...
 
 
+def _fruit_sweeter_than(statement: Select[tuple[Fruit]], value: int, **_ctx: Any) -> Select[tuple[Fruit]]:
+    return statement.where(Fruit.sweetness >= value)
+
+
+@strawchemy.filter(Fruit, include=["id", "name", "sweetness"], name="FruitFineGrainedFilter")
+class FruitFineGrainedFilter:
+    name: TextComparison = strawchemy.filter_field(ops=["eq", "like"])
+    sweeter_than: int = strawchemy.filter_field(apply=_fruit_sweeter_than)
+    sweeter_than_in: int = strawchemy.filter_field(apply=_fruit_sweeter_than, join="in")
+
+
 @strawchemy.order(Fruit, include="all", scope="schema")
 class FruitOrderBy: ...
 
@@ -322,6 +334,9 @@ class AsyncQuery:
     fruits: list[FruitType] = strawchemy.field(
         filter_input=FruitFilter, order_by_input=FruitOrderBy, repository_type=StrawchemyAsyncRepository
     )
+    fruits_fine_grained: list[FruitType] = strawchemy.field(
+        filter_input=FruitFineGrainedFilter, repository_type=StrawchemyAsyncRepository
+    )
     fruits_paginated: list[FruitTypeWithPaginationAndOrderBy] = strawchemy.field(
         filter_input=FruitFilter,
         pagination=True,
@@ -446,6 +461,9 @@ class SyncQuery:
     fruits: list[FruitType] = strawchemy.field(
         filter_input=FruitFilter, order_by_input=FruitOrderBy, repository_type=StrawchemySyncRepository
     )
+    fruits_fine_grained: list[FruitType] = strawchemy.field(
+        filter_input=FruitFineGrainedFilter, repository_type=StrawchemySyncRepository
+    )
     fruits_paginated: list[FruitTypeWithPaginationAndOrderBy] = strawchemy.field(
         filter_input=FruitFilter,
         pagination=True,
diff --git a/tests/integration/types/sqlite.py b/tests/integration/types/sqlite.py
index 405e8fe6..3cf0e898 100644
--- a/tests/integration/types/sqlite.py
+++ b/tests/integration/types/sqlite.py
@@ -17,6 +17,7 @@
     Strawchemy,
     StrawchemyAsyncRepository,
     StrawchemySyncRepository,
+    TextComparison,
     ValidationErrorType,
 )
 from strawchemy.schema.pagination import DefaultOffsetPagination
@@ -158,6 +159,17 @@ class FruitTypeWithPaginationAndOrderBy: ...
 class FruitFilter: ...
 
 
+def _fruit_sweeter_than(statement: Select[tuple[Fruit]], value: int, **_ctx: Any) -> Select[tuple[Fruit]]:
+    return statement.where(Fruit.sweetness >= value)
+
+
+@strawchemy.filter(Fruit, include=["id", "name", "sweetness"], name="FruitFineGrainedFilter")
+class FruitFineGrainedFilter:
+    name: TextComparison = strawchemy.filter_field(ops=["eq", "like"])
+    sweeter_than: int = strawchemy.filter_field(apply=_fruit_sweeter_than)
+    sweeter_than_in: int = strawchemy.filter_field(apply=_fruit_sweeter_than, join="in")
+
+
 @strawchemy.order(Fruit, include="all", scope="schema")
 class FruitOrderBy: ...
 
@@ -302,6 +314,9 @@ class AsyncQuery:
     fruits: list[FruitType] = strawchemy.field(
         filter_input=FruitFilter, order_by_input=FruitOrderBy, repository_type=StrawchemyAsyncRepository
     )
+    fruits_fine_grained: list[FruitType] = strawchemy.field(
+        filter_input=FruitFineGrainedFilter, repository_type=StrawchemyAsyncRepository
+    )
     fruits_paginated: list[FruitTypeWithPaginationAndOrderBy] = strawchemy.field(
         filter_input=FruitFilter,
         pagination=True,
@@ -426,6 +441,9 @@ class SyncQuery:
     fruits: list[FruitType] = strawchemy.field(
         filter_input=FruitFilter, order_by_input=FruitOrderBy, repository_type=StrawchemySyncRepository
     )
+    fruits_fine_grained: list[FruitType] = strawchemy.field(
+        filter_input=FruitFineGrainedFilter, repository_type=StrawchemySyncRepository
+    )
     fruits_paginated: list[FruitTypeWithPaginationAndOrderBy] = strawchemy.field(
         filter_input=FruitFilter,
         pagination=True,
diff --git a/tests/unit/__snapshots__/test_fine_grained_filters/test_fine_grained_schema.gql b/tests/unit/__snapshots__/test_fine_grained_filters/test_fine_grained_schema.gql
new file mode 100644
index 00000000..214415bb
--- /dev/null
+++ b/tests/unit/__snapshots__/test_fine_grained_filters/test_fine_grained_schema.gql
@@ -0,0 +1,75 @@
+'''
+"""Date with time (isoformat)"""
+scalar DateTime
+
+"""
+Boolean expression to compare DateTime fields. All fields are combined with logical 'AND'
+"""
+input DateTimeComparison {
+  eq: DateTime
+  neq: DateTime
+  isNull: Boolean
+  in: [DateTime!]
+  nin: [DateTime!]
+  gt: DateTime
+  gte: DateTime
+  lt: DateTime
+  lte: DateTime
+  year: IntOrderComparison
+  month: IntOrderComparison
+  day: IntOrderComparison
+  weekDay: IntOrderComparison
+  week: IntOrderComparison
+  quarter: IntOrderComparison
+  isoYear: IntOrderComparison
+  isoWeekDay: IntOrderComparison
+  hour: IntOrderComparison
+  minute: IntOrderComparison
+  second: IntOrderComparison
+}
+
+"""
+Boolean expression to compare fields supporting order comparisons. All fields are combined with logical 'AND'
+"""
+input IntOrderComparison {
+  eq: Int
+  neq: Int
+  isNull: Boolean
+  in: [Int!]
+  nin: [Int!]
+  gt: Int
+  gte: Int
+  lt: Int
+  lte: Int
+}
+
+type Query {
+  """Fetch objects from the TicketType collection"""
+  tickets(filter: TicketFilter = null): [TicketType!]!
+}
+
+"""TextComparison restricted to eq, in, like"""
+input TextComparisonStrEqInLike {
+  eq: String
+  in: [String!]
+  like: String
+}
+
+"""
+Boolean expression to compare fields. All fields are combined with logical 'AND'.
+"""
+input TicketFilter {
+  _and: [TicketFilter!]! = []
+  _or: [TicketFilter!]! = []
+  _not: TicketFilter
+  name: TextComparisonStrEqInLike
+  publishedAt: DateTimeComparison
+  publishedAfter: DateTime
+}
+
+"""GraphQL type"""
+type TicketType {
+  name: String!
+  publishedAt: DateTime!
+}
+'''
\ No newline at end of file
diff --git a/tests/unit/test_fine_grained_filters.py b/tests/unit/test_fine_grained_filters.py
new file mode 100644
index 00000000..1338fb25
--- /dev/null
+++ b/tests/unit/test_fine_grained_filters.py
@@ -0,0 +1,315 @@
+from __future__ import annotations
+
+from datetime import datetime
+from datetime import datetime as _dt
+from typing import TYPE_CHECKING, Any
+
+import pytest
+import strawberry
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+from strawberry import UNSET
+from strawberry.types import get_object_definition
+
+from strawchemy import Strawchemy
+from strawchemy.dto.strawberry import CustomFilter, CustomFilterFieldDefinition, Filter
+from strawchemy.exceptions import StrawchemyFieldError
+from strawchemy.schema.filters.fields import FilterFieldMarker
+from strawchemy.schema.filters.inputs import OrderComparison, TextComparison
+
+if TYPE_CHECKING:
+    from syrupy.assertion import SnapshotAssertion
+
+# Module-level convenience: `filter_field` is a Strawchemy method, but markers are
+# mapper-agnostic, so a shared alias keeps the rest of the tests terse. Tests that
+# specifically exercise the method call `Strawchemy("sqlite").filter_field(...)` directly.
+_module_strawchemy = Strawchemy("sqlite")
+filter_field = _module_strawchemy.filter_field
+
+
+class _Base(DeclarativeBase): ...
+
+
+class _Ticket(_Base):
+    __tablename__ = "fgf_ticket"
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column()
+    published_at: Mapped[datetime] = mapped_column()
+
+
+def test_filter_field_bare_returns_marker() -> None:
+    marker = Strawchemy("sqlite").filter_field()
+    assert isinstance(marker, FilterFieldMarker)
+    assert marker.ops is None
+    assert marker.apply is None
+
+
+def test_filter_field_with_ops_normalizes_to_tuple() -> None:
+    marker = Strawchemy("sqlite").filter_field(ops=["eq", "in", "like"])
+    assert marker.ops == ("eq", "in", "like")
+    assert marker.apply is None
+
+
+def test_filter_field_with_apply_captures_callable_and_join() -> None:
+    def _apply(statement: Any, *_args: Any, **_kwargs: Any) -> Any:
+        return statement
+
+    marker = Strawchemy("sqlite").filter_field(apply=_apply, join="in")
+    assert marker.apply is _apply
+    assert marker.join == "in"
+    assert marker.ops is None
+
+
+def test_filter_field_rejects_ops_and_apply_together() -> None:
+    with pytest.raises(StrawchemyFieldError, match="mutually exclusive"):
+        Strawchemy("sqlite").filter_field(ops=["eq"], apply=lambda s, _v, **_k: s)
+
+
+def test_filter_field_rejects_unknown_join() -> None:
+    with pytest.raises(StrawchemyFieldError, match="join"):
+        Strawchemy("sqlite").filter_field(apply=lambda s, _v, **_k: s, join="bogus")  # ty: ignore[invalid-argument-type]
+
+
+def test_restricted_exposes_only_selected_ops() -> None:
+    cls = TextComparison.restricted(str, ("eq", "in", "like"))
+    exposed = {f.graphql_name or f.name for f in get_object_definition(cls, strict=True).fields}
+    assert exposed == {"eq", "in", "like"}
+
+
+def test_restricted_keeps_unselected_attrs_unset_at_runtime() -> None:
+    cls = TextComparison.restricted(str, ("eq",))
+    instance = cls()
+    assert instance.gt is UNSET  # unselected op still present (UNSET) so query-time filters work
+    assert instance.like is UNSET
+
+
+def test_restricted_carries_base_filter() -> None:
+    cls = TextComparison.restricted(str, ("eq",))
+    assert cls.__strawchemy_comparison__.filter is TextComparison.__strawchemy_comparison__.filter
+
+
+def test_restricted_invalid_op_raises() -> None:
+    with pytest.raises(StrawchemyFieldError, match="frobnicate"):
+        TextComparison.restricted(str, ("eq", "frobnicate"))
+
+
+def test_restricted_is_deduped() -> None:
+    a = TextComparison.restricted(str, ("eq", "like"))
+    b = TextComparison.restricted(str, ("like", "eq"))  # order-independent
+    assert a is b
+
+
+def test_restricted_over_int_carries_int() -> None:
+    cls = OrderComparison.restricted(int, ("eq", "gt"))
+    eq_field = next(f for f in get_object_definition(cls, strict=True).fields if f.name == "eq")
+    eq_type = eq_field.type
+    # strawberry wraps `int | None` as a StrawberryOptional whose `of_type` is the scalar
+    eq_type = getattr(eq_type, "of_type", eq_type)
+    assert int in getattr(eq_type, "__args__", (eq_type,))
+
+
+def test_custom_filter_is_filter_member() -> None:
+    cf = CustomFilter(apply=lambda s, _v, **_k: s, value=5, join="exists", field_node=None)  # ty: ignore[invalid-argument-type]
+    f = Filter(and_=[cf])
+    assert f.and_ == [cf]
+
+
+def test_custom_filter_field_definition_carries_marker_data() -> None:
+    from datetime import datetime
+
+    from strawchemy.dto.types import DTOConfig, Purpose
+
+    def _apply(statement: Any, _value: Any, **_ctx: Any) -> Any:
+        return statement
+
+    field_def = CustomFilterFieldDefinition(
+        dto_config=DTOConfig(Purpose.READ),
+        model=object,  # ty: ignore[invalid-argument-type]
+        model_field_name="published_after",
+        type_hint=datetime,
+        apply=_apply,
+        join="exists",
+    )
+    assert field_def.apply is _apply
+    assert field_def.join == "exists"
+
+
+def test_parse_declared_fields_collects_restricted_and_custom() -> None:
+    sc = Strawchemy("sqlite")
+
+    def _apply(statement: Any, _value: Any, **_ctx: Any) -> Any:
+        return statement
+
+    class TicketFilter:
+        name: str = sc.filter_field(ops=["eq", "in", "like"])
+        published_after: _dt = sc.filter_field(apply=_apply)
+
+    declared = sc.filter_factory.parse_declared_filter_fields(TicketFilter)
+    assert set(declared) == {"name", "published_after"}
+
+    name_marker, name_annotation = declared["name"]
+    assert name_marker.ops == ("eq", "in", "like")
+    assert name_annotation is str  # data type from the annotation
+
+    custom_marker, custom_annotation = declared["published_after"]
+    assert custom_marker.apply is _apply
+    assert custom_annotation is _dt
+
+
+def test_parse_declared_fields_requires_annotation() -> None:
+    sc = Strawchemy("sqlite")
+
+    class Bad:
+        # restricted field with no annotation -> no data type to build a comparison
+        name = sc.filter_field(ops=["eq"])
+
+    with pytest.raises(StrawchemyFieldError, match="annotation"):
+        sc.filter_factory.parse_declared_filter_fields(Bad)
+
+
+def test_restricted_field_overrides_auto_comparison() -> None:
+    strawchemy = Strawchemy("sqlite")
+
+    @strawchemy.filter(_Ticket, include=["name"])
+    class TicketFilter:
+        # The annotation is the comparison type (documentary); the data type comes from the column.
+        name: TextComparison = filter_field(ops=["eq", "like"])
+
+    definition = get_object_definition(TicketFilter, strict=True)
+    name_field = next(f for f in definition.fields if (f.graphql_name or f.name) == "name")
+    comparison = name_field.type
+    # strawberry wraps `Comparison | None` as a StrawberryOptional whose `of_type` is the comparison
+    comparison_origin = getattr(comparison, "of_type", comparison)
+    inner = get_object_definition(comparison_origin, strict=True)
+    exposed = {f.graphql_name or f.name for f in inner.fields}
+    assert exposed == {"eq", "like"}
+
+
+def test_filter_field_forwards_strawberry_field_args() -> None:
+    """filter_field()'s strawberry.field kwargs reach the generated GraphQL input field."""
+    strawchemy = Strawchemy("sqlite")
+
+    @strawchemy.filter(_Ticket, include=["name"])
+    class TicketFilter:
+        name: TextComparison = filter_field(
+            ops=["eq"],
+            name="renamedName",
+            description="Filter by ticket name",
+            deprecation_reason="use something else",
+            metadata={"k": "v"},
+        )
+
+    definition = get_object_definition(TicketFilter, strict=True)
+    field = next(f for f in definition.fields if (f.graphql_name or f.name) == "renamedName")
+    assert field.description == "Filter by ticket name"
+    assert field.deprecation_reason == "use something else"
+    assert field.metadata == {"k": "v"}
+
+
+def test_restricted_field_comparison_annotation_mismatch_raises() -> None:
+    strawchemy = Strawchemy("sqlite")
+
+    with pytest.raises(StrawchemyFieldError, match="OrderComparison"):
+
+        @strawchemy.filter(_Ticket, include=["name"])
+        class TicketFilter:
+            # `name` is a str column -> TextComparison; OrderComparison is the wrong shape.
+            name: OrderComparison = filter_field(ops=["eq"])
+
+
+def test_custom_apply_field_injected_as_scalar() -> None:
+    strawchemy = Strawchemy("sqlite")
+
+    def _published_after(statement, value, **_ctx):  # noqa: ANN001, ANN003, ANN202
+        return statement.where(_Ticket.published_at >= value)
+
+    @strawchemy.filter(_Ticket, include=["name"])
+    class TicketFilter:
+        published_after: datetime = filter_field(apply=_published_after)
+
+    definition = get_object_definition(TicketFilter, strict=True)
+    # camelCase is applied at schema build time; the object definition keeps the python name
+    field = next((f for f in definition.fields if (f.graphql_name or f.name) == "published_after"), None)
+    assert field is not None
+
+
+def test_filters_tree_wraps_custom_field() -> None:
+    strawchemy = Strawchemy("sqlite")
+
+    def _published_after(statement, value, **_ctx):  # noqa: ANN001, ANN003, ANN202
+        return statement.where(_Ticket.published_at >= value)
+
+    @strawchemy.filter(_Ticket, include=["name"])
+    class TicketFilter:
+        published_after: datetime = filter_field(apply=_published_after)
+
+    instance = TicketFilter()
+    instance.published_after = datetime(2024, 1, 1)  # ty: ignore[unresolved-attribute]  # noqa: DTZ001
+    _node, query = instance.filters_tree()
+
+    assert any(isinstance(item, CustomFilter) for item in query.and_)
+    custom = next(item for item in query.and_ if isinstance(item, CustomFilter))
+    assert custom.apply is _published_after
+    assert custom.value == datetime(2024, 1, 1)  # noqa: DTZ001
+
+
+def test_public_surface_end_to_end() -> None:
+    sc = Strawchemy("sqlite")
+    assert callable(sc.filter_field)
+
+    def _published_after(statement, value, **_ctx):  # noqa: ANN001, ANN003, ANN202
+        return statement.where(_Ticket.published_at >= value)
+
+    @sc.filter(_Ticket, include=["name"])
+    class TicketFilter:
+        name: str = sc.filter_field(ops=["eq"])
+        published_after: datetime = sc.filter_field(apply=_published_after)
+
+    definition = get_object_definition(TicketFilter, strict=True)
+    exposed = {f.graphql_name or f.name for f in definition.fields}
+    assert "name" in exposed
+    assert "published_after" in exposed
+
+
+def test_restricted_field_on_unknown_column_raises() -> None:
+    from strawchemy.exceptions import StrawchemyFieldError
+
+    sc = Strawchemy("sqlite")
+    with pytest.raises(StrawchemyFieldError, match="nonexistent"):
+
+        @sc.filter(_Ticket, include=["name"])
+        class TicketFilter:
+            nonexistent: str = filter_field(ops=["eq"])
+
+
+def test_invalid_operator_raises_at_definition() -> None:
+    from strawchemy.exceptions import StrawchemyFieldError
+
+    sc = Strawchemy("sqlite")
+    with pytest.raises(StrawchemyFieldError, match="frobnicate"):
+
+        @sc.filter(_Ticket, include=["name"])
+        class TicketFilter:
+            name: str = filter_field(ops=["frobnicate"])
+
+
+@pytest.mark.snapshot
+def test_fine_grained_schema(graphql_snapshot: SnapshotAssertion) -> None:
+    sc = Strawchemy("sqlite")
+
+    def _published_after(statement: Any, value: Any, **_ctx: Any) -> Any:
+        return statement.where(_Ticket.published_at >= value)
+
+    @sc.type(_Ticket, include=["name", "published_at"])
+    class TicketType: ...
+
+    @sc.filter(_Ticket, include=["name", "published_at"])
+    class TicketFilter:
+        name: TextComparison = filter_field(ops=["eq", "in", "like"])
+        published_after: datetime = filter_field(apply=_published_after)
+
+    @strawberry.type
+    class Query:
+        tickets: list[TicketType] = sc.field(filter_input=TicketFilter)
+
+    schema = strawberry.Schema(query=Query)
+    assert str(schema) == graphql_snapshot