Repository: deepset-ai/haystack
Pull request: feat: Implement apply_filter_policy and FilterPolicy.MERGE for the new filters
### Why:
Implements proper merging of new filters.

- fixes: https://github.com/deepset-ai/haystack/issues/7995

### What:
- Implemented new utility functions to determine filter types (`is_legacy`, `is_comparison_filter`, `is_logical_filter`).
- Extended the `apply_filter: policy` function to support merging of `init_filters` with `runtime_filters` using a logical operator, facilitating complex filter scenarios.
- Added a comprehensive suite of tests to validate the functionality of applying filter policies including scenarios with no filters, comparison filters, logical operators, and user-defined logical operators.

### How can it be used:
The enhancements enable complex filter logic to be applied seamlessly in document query operations, such as:
- Merging initial and runtime filters with support for legacy filter formats.
- Applying logical operators (`AND`, `OR`) when merging comparison or logical filters, allowing for a more nuanced filter logic that can accurately reflect user needs or query specifics.

Example usage of merging comparison and logical filters:
```python
init_filter = {"field": "meta.type", "operator": "==", "value": "pdf"}
runtime_filter = {
    "operator": "AND",
    "conditions": [
        {"field": "meta.name", "operator": "==", "value": "John"},
        {"field": "meta.year", "operator": "==", "value": "2022"},
    ],
}
# Merging the above would result in runtime_filter including the init_filter as another condition under the same logical operator "AND".
```

### How did you test it:
A series of unit tests were added covering various scenarios including:
- Merging filters under both `MERGE` and `REPLACE` policies with different combinations of comparison and logical filters.
- Ensuring that the correct logical operator is applied during the merge process.
- Testing the behavior with no filters provided, ensuring backward compatibility and robust error handling.

### Notes for the reviewer:
- Special attention should be given to the logic involving merging different types of filters (comparison vs. logical) and ensuring the intended behavior aligns with real-world use cases.
- Review the test cases for `apply_filter_policy` to ensure all potential scenarios are covered and the expected behavior is clearly documented and verified.

Problem statement:
`apply_filter_policy` and `FilterPolicy.MERGE` don't work properly for the new filter syntax
**Is your feature request related to a problem? Please describe.**
The current `apply_filter_policy` function in our retrieval system only supports legacy dictionary-based filters. This limitation becomes problematic for new filters syntax. For example, we can't merge filters like: 
`{"field": "meta.name", "operator": "==", "value": "John"}` and 
`{"field": "meta.year", "operator": "==", "value": "2022"}` using the current implementation.

**Describe the solution you'd like**
Update the `apply_filter_policy` function to support both the legacy filters and a new filter format merge functionality. The function should:

Accept both old and new filters as input for `init_filters` and `runtime_filters`.
Properly merge (or replace) filters based on the `FilterPolicy` for both formats.

**Describe alternatives you've considered**
Not considered

**Additional context**
This change is crucial for improving handling both legacy and new filters and applying `FilterPolicy` correctly.

Hints from issue discussion:
@julian-risch @mrm1001 this one is needed for the final 2.3.0 release cc @sjrl 
I have the most context to fix this one and @sjrl will review
So I think there are three basic scenarios that we might like to cover:

1. Combining two comparison filters together where a logical operator needs to be provided by the user. For example,
```python
comparison_filter_1 = {"field": "meta.name", "operator": "==", "value": "John"}
comparison_filter_2 = {"field": "meta.year", "operator": "==", "value": "2022"}
user_provided_logical_filter = "AND"
combined_filter = {
    "operator": "AND",
    "conditions": [
      {"field": "meta.name", "operator": "==", "value": "John"},
      {"field": "meta.year", "operator": "==", "value": "2022"}
    ]
}
```
* An edge case we should consider for this scenario is what we do if both filters have the same field. For example, if the fields are both `meta.name` which one should be used over the other. 

2. Combining a comparison filter and a logical filter
  * I think in this scenario a user provided logical filter should still be provided. Perhaps if it matches the existing logical operator we should merge them, but I'm not entirely sure how we should handle when they differ.
```python
comparison_filter = {"field": "meta.type", "operator": "==", "value": "pdf"}
logical_filter = {
    "operator": "AND",
    "conditions": [
      {"field": "meta.name", "operator": "==", "value": "John"},
      {"field": "meta.year", "operator": "==", "value": "2022"}
    ]
}
user_provided_logical_filter = "AND"
combined_filter = {
    "operator": "AND",
    "conditions": [
      {"field": "meta.name", "operator": "==", "value": "John"},
      {"field": "meta.year", "operator": "==", "value": "2022"},
      {"field": "meta.type", "operator": "==", "value": "pdf"}
    ]
}
```
3. Combining two logical filters
  * I'm less certain how to fully handle all situations in this case. I believe if both logical filters have the "AND" operator it should be fine to merge both of them. However, if one is a "OR" and the other and "AND" I'm not certain how they should be combined, some advice here would be greatly appreciated.
```python
logical_filter_1 = {
    "operator": "AND",
    "conditions": [
      {"field": "meta.type", "operator": "==", "value": "pdf"},
      {"field": "meta.month", "operator": "==", "value": "April"}
    ]
}
logical_filter_2 = {
    "operator": "AND",
    "conditions": [
      {"field": "meta.name", "operator": "==", "value": "John"},
      {"field": "meta.year", "operator": "==", "value": "2022"}
    ]
}
combined_filter = {
    "operator": "AND",
    "conditions": [
      {"field": "meta.name", "operator": "==", "value": "John"},
      {"field": "meta.year", "operator": "==", "value": "2022"},
      {"field": "meta.type", "operator": "==", "value": "pdf"},
      {"field": "meta.month", "operator": "==", "value": "April"}
    ]
}
```

Commit message: Add reno note

Patch to review:
diff --git a/haystack/document_stores/types/filter_policy.py b/haystack/document_stores/types/filter_policy.py
index a2be576d20..66f849adaa 100644
--- a/haystack/document_stores/types/filter_policy.py
+++ b/haystack/document_stores/types/filter_policy.py
@@ -36,10 +36,41 @@ def from_str(filter_policy: str) -> "FilterPolicy":
         return policy
 
 
+def is_legacy(filter_item: Dict[str, Any]) -> bool:
+    """
+    Check if the given filter is a legacy filter.
+
+    :param filter_item: The filter to check.
+    :returns: True if the filter is a legacy filter, False otherwise.
+    """
+    return "operator" not in filter_item
+
+
+def is_comparison_filter(filter_item: Dict[str, Any]) -> bool:
+    """
+    Check if the given filter is a comparison filter.
+
+    :param filter_item: The filter to check.
+    :returns: True if the filter is a comparison filter, False otherwise.
+    """
+    return all(key in filter_item for key in ["field", "operator", "value"])
+
+
+def is_logical_filter(filter_item: Dict[str, Any]) -> bool:
+    """
+    Check if the given filter is a logical filter.
+
+    :param filter_item: The filter to check.
+    :returns: True if the filter is a logical filter, False otherwise.
+    """
+    return "operator" in filter_item and "conditions" in filter_item
+
+
 def apply_filter_policy(
     filter_policy: FilterPolicy,
     init_filters: Optional[Dict[str, Any]] = None,
     runtime_filters: Optional[Dict[str, Any]] = None,
+    default_logical_operator: str = "AND",
 ) -> Optional[Dict[str, Any]]:
     """
     Apply the filter policy to the given initial and runtime filters to determine the final set of filters used.
@@ -52,10 +83,48 @@ def apply_filter_policy(
           values from the runtime filters will overwrite those from the initial filters.
     :param init_filters: The initial filters set during the initialization of the relevant retriever.
     :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
-                            can change for each query/retreiver run invocation.
+                            can change for each query/retriever run invocation.
+    :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
     :returns: A dictionary containing the resulting filters based on the provided policy.
     """
     if filter_policy == FilterPolicy.MERGE and runtime_filters:
-        return {**(init_filters or {}), **runtime_filters}
-    else:
-        return runtime_filters or init_filters
+        # legacy filters merge handling
+        if is_legacy(runtime_filters):
+            return {**(init_filters or {}), **runtime_filters}
+        elif init_filters is not None:
+            # here we merge new filters
+            def merge_comparison_filters(
+                filter1: Dict[str, Any], filter2: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if filter1["field"] == filter2["field"]:
+                    # When fields are the same, use the runtime filter (filter2)
+                    return filter2
+                return {"operator": logical_op, "conditions": [filter1, filter2]}
+
+            def merge_comparison_and_logical(
+                comparison: Dict[str, Any], logical: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if logical["operator"] == logical_op:
+                    logical["conditions"].append(comparison)
+                    return logical
+                else:
+                    return {"operator": logical_op, "conditions": [comparison, logical]}
+
+            def merge_logical_filters(
+                filter1: Dict[str, Any], filter2: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if filter1["operator"] == filter2["operator"] == logical_op:
+                    return {"operator": logical_op, "conditions": filter1["conditions"] + filter2["conditions"]}
+                else:
+                    return {"operator": logical_op, "conditions": [filter1, filter2]}
+
+            if is_comparison_filter(init_filters) and is_comparison_filter(runtime_filters):
+                return merge_comparison_filters(init_filters, runtime_filters, default_logical_operator)
+            elif is_comparison_filter(init_filters) and is_logical_filter(runtime_filters):
+                return merge_comparison_and_logical(init_filters, runtime_filters, default_logical_operator)
+            elif is_logical_filter(init_filters) and is_comparison_filter(runtime_filters):
+                return merge_comparison_and_logical(runtime_filters, init_filters, default_logical_operator)
+            elif is_logical_filter(init_filters) and is_logical_filter(runtime_filters):
+                return merge_logical_filters(init_filters, runtime_filters, default_logical_operator)
+
+    return runtime_filters or init_filters
diff --git a/releasenotes/notes/implement-merge-filter-logic-99e6785a78f80ae9.yaml b/releasenotes/notes/implement-merge-filter-logic-99e6785a78f80ae9.yaml
new file mode 100644
index 0000000000..32cbbca67e
--- /dev/null
+++ b/releasenotes/notes/implement-merge-filter-logic-99e6785a78f80ae9.yaml
@@ -0,0 +1,4 @@
+---
+enhancements:
+ - |
+   Enhanced filter application logic to support merging of filters. It facilitates more precise retrieval filtering, allowing for both init and runtime complex filter combinations with logical operators.
diff --git a/test/utils/test_filters.py b/test/utils/test_filters.py
index eccf1ca1ec..edb8aacbd2 100644
--- a/test/utils/test_filters.py
+++ b/test/utils/test_filters.py
@@ -3,6 +3,9 @@
 # SPDX-License-Identifier: Apache-2.0
 import pytest
 import pandas as pd
+from haystack.document_stores.types import FilterPolicy
+
+from haystack.document_stores.types.filter_policy import apply_filter_policy
 
 from haystack import Document
 from haystack.errors import FilterError
@@ -726,3 +729,122 @@ def test_convert_with_incorrect_filter_nesting():
 
     with pytest.raises(FilterError):
         convert({"number": {"page": {"chapter": "intro"}}})
+
+
+################################################################################
+# Tests for the apply_filter_policy function
+################################################################################
+def test_no_filters():
+    assert apply_filter_policy(FilterPolicy.MERGE, None, None) is None
+
+
+def test_replace_policy():
+    init_filter = {"field": "meta.name", "operator": "==", "value": "John"}
+    runtime_filter = {"field": "meta.year", "operator": "==", "value": "2022"}
+    assert apply_filter_policy(FilterPolicy.REPLACE, init_filter, runtime_filter) == runtime_filter
+
+
+def test_merge_comparison_filters():
+    init_filter = {"field": "meta.name", "operator": "==", "value": "John"}
+    runtime_filter = {"field": "meta.year", "operator": "==", "value": "2022"}
+    expected = {"operator": "AND", "conditions": [init_filter, runtime_filter]}
+    assert apply_filter_policy(FilterPolicy.MERGE, init_filter, runtime_filter) == expected
+
+
+def test_merge_comparison_filters_same_field():
+    init_filter = {"field": "meta.name", "operator": "==", "value": "John"}
+    runtime_filter = {"field": "meta.name", "operator": "==", "value": "Jane"}
+    assert apply_filter_policy(FilterPolicy.MERGE, init_filter, runtime_filter) == runtime_filter
+
+
+def test_merge_comparison_and_logical():
+    comparison_filter = {"field": "meta.type", "operator": "==", "value": "pdf"}
+    logical_filter = {
+        "operator": "AND",
+        "conditions": [
+            {"field": "meta.name", "operator": "==", "value": "John"},
+            {"field": "meta.year", "operator": "==", "value": "2022"},
+        ],
+    }
+    expected = {
+        "operator": "AND",
+        "conditions": [
+            {"field": "meta.name", "operator": "==", "value": "John"},
+            {"field": "meta.year", "operator": "==", "value": "2022"},
+            {"field": "meta.type", "operator": "==", "value": "pdf"},
+        ],
+    }
+    assert apply_filter_policy(FilterPolicy.MERGE, comparison_filter, logical_filter) == expected
+
+
+def test_merge_comparison_and_logical_different_operator():
+    comparison_filter = {"field": "meta.type", "operator": "==", "value": "pdf"}
+    logical_filter = {
+        "operator": "OR",
+        "conditions": [
+            {"field": "meta.name", "operator": "==", "value": "John"},
+            {"field": "meta.year", "operator": "==", "value": "2022"},
+        ],
+    }
+    expected = {"operator": "AND", "conditions": [comparison_filter, logical_filter]}
+    assert apply_filter_policy(FilterPolicy.MERGE, comparison_filter, logical_filter) == expected
+
+
+def test_merge_logical_filters_same_operator():
+    logical_filter1 = {
+        "operator": "AND",
+        "conditions": [
+            {"field": "meta.type", "operator": "==", "value": "pdf"},
+            {"field": "meta.month", "operator": "==", "value": "April"},
+        ],
+    }
+    logical_filter2 = {
+        "operator": "AND",
+        "conditions": [
+            {"field": "meta.name", "operator": "==", "value": "John"},
+            {"field": "meta.year", "operator": "==", "value": "2022"},
+        ],
+    }
+    expected = {
+        "operator": "AND",
+        "conditions": [
+            {"field": "meta.type", "operator": "==", "value": "pdf"},
+            {"field": "meta.month", "operator": "==", "value": "April"},
+            {"field": "meta.name", "operator": "==", "value": "John"},
+            {"field": "meta.year", "operator": "==", "value": "2022"},
+        ],
+    }
+    assert apply_filter_policy(FilterPolicy.MERGE, logical_filter1, logical_filter2) == expected
+
+
+def test_merge_logical_filters_different_operator():
+    logical_filter1 = {
+        "operator": "AND",
+        "conditions": [
+            {"field": "meta.type", "operator": "==", "value": "pdf"},
+            {"field": "meta.month", "operator": "==", "value": "April"},
+        ],
+    }
+    logical_filter2 = {
+        "operator": "OR",
+        "conditions": [
+            {"field": "meta.name", "operator": "==", "value": "John"},
+            {"field": "meta.year", "operator": "==", "value": "2022"},
+        ],
+    }
+    expected = {"operator": "AND", "conditions": [logical_filter1, logical_filter2]}
+    assert apply_filter_policy(FilterPolicy.MERGE, logical_filter1, logical_filter2) == expected
+
+
+def test_user_provided_logical_operator():
+    init_filter = {"field": "meta.name", "operator": "==", "value": "John"}
+    runtime_filter = {"field": "meta.year", "operator": "==", "value": "2022"}
+    expected = {"operator": "OR", "conditions": [init_filter, runtime_filter]}
+    assert apply_filter_policy(FilterPolicy.MERGE, init_filter, runtime_filter, "OR") == expected
+
+
+def test_default_logical_operator():
+    init_filter = {"field": "meta.name", "operator": "==", "value": "John"}
+    runtime_filter = {"field": "meta.year", "operator": "==", "value": "2022"}
+    expected = {"operator": "AND", "conditions": [init_filter, runtime_filter]}
+    assert apply_filter_policy(FilterPolicy.MERGE, init_filter, runtime_filter) == expected


Code review comments:
diff_hunk: @@ -36,10 +36,41 @@ def from_str(filter_policy: str) -> "FilterPolicy":
         return policy
 
 
+def is_legacy(filter_item: Dict[str, Any]) -> bool:
+    """
+    Check if the given filter is a legacy filter.
+
+    :param filter_item: The filter to check.
+    :returns: True if the filter is a legacy filter, False otherwise.
+    """
+    return "operator" not in filter_item
+
+
+def is_comparison_filter(filter_item: Dict[str, Any]) -> bool:
+    """
+    Check if the given filter is a comparison filter.
+
+    :param filter_item: The filter to check.
+    :returns: True if the filter is a comparison filter, False otherwise.
+    """
+    return all(key in filter_item for key in ["field", "operator", "value"])
+
+
+def is_logical_filter(filter_item: Dict[str, Any]) -> bool:
+    """
+    Check if the given filter is a logical filter.
+
+    :param filter_item: The filter to check.
+    :returns: True if the filter is a logical filter, False otherwise.
+    """
+    return "operator" in filter_item and "conditions" in filter_item
+
+
 def apply_filter_policy(
     filter_policy: FilterPolicy,
     init_filters: Optional[Dict[str, Any]] = None,
     runtime_filters: Optional[Dict[str, Any]] = None,
+    default_logical_operator: str = "AND",
path: haystack/document_stores/types/filter_policy.py
text: @user1:
Missing doc string. Also do we know what are possible values this can be already, if so could we make this an Enum or a Literal type?

diff_hunk: @@ -52,10 +83,48 @@ def apply_filter_policy(
           values from the runtime filters will overwrite those from the initial filters.
     :param init_filters: The initial filters set during the initialization of the relevant retriever.
     :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
-                            can change for each query/retreiver run invocation.
+                            can change for each query/retriever run invocation.
+    :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
     :returns: A dictionary containing the resulting filters based on the provided policy.
     """
     if filter_policy == FilterPolicy.MERGE and runtime_filters:
-        return {**(init_filters or {}), **runtime_filters}
-    else:
-        return runtime_filters or init_filters
+        # legacy filters merge handling
+        if is_legacy(runtime_filters):
+            return {**(init_filters or {}), **runtime_filters}
+        elif init_filters is not None:
+            # here we merge new filters
+            def merge_comparison_filters(
+                filter1: Dict[str, Any], filter2: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if filter1["field"] == filter2["field"]:
+                    # When fields are the same, use the runtime filter (filter2)
+                    return filter2
+                return {"operator": logical_op, "conditions": [filter1, filter2]}
+
+            def merge_comparison_and_logical(
+                comparison: Dict[str, Any], logical: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if logical["operator"] == logical_op:
+                    logical["conditions"].append(comparison)
+                    return logical
+                else:
+                    return {"operator": logical_op, "conditions": [comparison, logical]}
+
+            def merge_logical_filters(
+                filter1: Dict[str, Any], filter2: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if filter1["operator"] == filter2["operator"] == logical_op:
+                    return {"operator": logical_op, "conditions": filter1["conditions"] + filter2["conditions"]}
path: haystack/document_stores/types/filter_policy.py
text: @user1:
This doesn’t handle if the conditions contain filters on the same field. This could cause an invalid filter to be made.

diff_hunk: @@ -52,10 +83,48 @@ def apply_filter_policy(
           values from the runtime filters will overwrite those from the initial filters.
     :param init_filters: The initial filters set during the initialization of the relevant retriever.
     :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
-                            can change for each query/retreiver run invocation.
+                            can change for each query/retriever run invocation.
+    :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
     :returns: A dictionary containing the resulting filters based on the provided policy.
     """
     if filter_policy == FilterPolicy.MERGE and runtime_filters:
-        return {**(init_filters or {}), **runtime_filters}
-    else:
-        return runtime_filters or init_filters
+        # legacy filters merge handling
+        if is_legacy(runtime_filters):
+            return {**(init_filters or {}), **runtime_filters}
+        elif init_filters is not None:
+            # here we merge new filters
+            def merge_comparison_filters(
+                filter1: Dict[str, Any], filter2: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if filter1["field"] == filter2["field"]:
+                    # When fields are the same, use the runtime filter (filter2)
+                    return filter2
+                return {"operator": logical_op, "conditions": [filter1, filter2]}
+
+            def merge_comparison_and_logical(
+                comparison: Dict[str, Any], logical: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if logical["operator"] == logical_op:
+                    logical["conditions"].append(comparison)
+                    return logical
+                else:
+                    return {"operator": logical_op, "conditions": [comparison, logical]}
+
+            def merge_logical_filters(
+                filter1: Dict[str, Any], filter2: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if filter1["operator"] == filter2["operator"] == logical_op:
+                    return {"operator": logical_op, "conditions": filter1["conditions"] + filter2["conditions"]}
+                else:
+                    return {"operator": logical_op, "conditions": [filter1, filter2]}
path: haystack/document_stores/types/filter_policy.py
text: @user1:
Same problem as above we shouldn’t merge in this way.

diff_hunk: @@ -52,10 +83,48 @@ def apply_filter_policy(
           values from the runtime filters will overwrite those from the initial filters.
     :param init_filters: The initial filters set during the initialization of the relevant retriever.
     :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
-                            can change for each query/retreiver run invocation.
+                            can change for each query/retriever run invocation.
+    :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
     :returns: A dictionary containing the resulting filters based on the provided policy.
     """
     if filter_policy == FilterPolicy.MERGE and runtime_filters:
-        return {**(init_filters or {}), **runtime_filters}
-    else:
-        return runtime_filters or init_filters
+        # legacy filters merge handling
+        if is_legacy(runtime_filters):
+            return {**(init_filters or {}), **runtime_filters}
+        elif init_filters is not None:
+            # here we merge new filters
+            def merge_comparison_filters(
path: haystack/document_stores/types/filter_policy.py
text: @user1:
Can we pull out these functions and make them standalone. It’s kind of odd to define them nested within another function. And they could be useful utility functions for users.

diff_hunk: @@ -52,10 +83,48 @@ def apply_filter_policy(
           values from the runtime filters will overwrite those from the initial filters.
     :param init_filters: The initial filters set during the initialization of the relevant retriever.
     :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
-                            can change for each query/retreiver run invocation.
+                            can change for each query/retriever run invocation.
+    :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
     :returns: A dictionary containing the resulting filters based on the provided policy.
     """
     if filter_policy == FilterPolicy.MERGE and runtime_filters:
-        return {**(init_filters or {}), **runtime_filters}
-    else:
-        return runtime_filters or init_filters
+        # legacy filters merge handling
+        if is_legacy(runtime_filters):
+            return {**(init_filters or {}), **runtime_filters}
+        elif init_filters is not None:
+            # here we merge new filters
+            def merge_comparison_filters(
+                filter1: Dict[str, Any], filter2: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if filter1["field"] == filter2["field"]:
+                    # When fields are the same, use the runtime filter (filter2)
path: haystack/document_stores/types/filter_policy.py
text: @user1:
For these cases where we make a decision to overwrite let’s add a logger message alerting the user that this happening.

diff_hunk: @@ -52,10 +83,48 @@ def apply_filter_policy(
           values from the runtime filters will overwrite those from the initial filters.
     :param init_filters: The initial filters set during the initialization of the relevant retriever.
     :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
-                            can change for each query/retreiver run invocation.
+                            can change for each query/retriever run invocation.
+    :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
     :returns: A dictionary containing the resulting filters based on the provided policy.
     """
     if filter_policy == FilterPolicy.MERGE and runtime_filters:
-        return {**(init_filters or {}), **runtime_filters}
-    else:
-        return runtime_filters or init_filters
+        # legacy filters merge handling
+        if is_legacy(runtime_filters):
+            return {**(init_filters or {}), **runtime_filters}
+        elif init_filters is not None:
+            # here we merge new filters
+            def merge_comparison_filters(
+                filter1: Dict[str, Any], filter2: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if filter1["field"] == filter2["field"]:
+                    # When fields are the same, use the runtime filter (filter2)
+                    return filter2
+                return {"operator": logical_op, "conditions": [filter1, filter2]}
+
+            def merge_comparison_and_logical(
+                comparison: Dict[str, Any], logical: Dict[str, Any], logical_op: str
+            ) -> Dict[str, Any]:
+                if logical["operator"] == logical_op:
+                    logical["conditions"].append(comparison)
+                    return logical
+                else:
+                    return {"operator": logical_op, "conditions": [comparison, logical]}
path: haystack/document_stores/types/filter_policy.py
text: @user1:
I don’t like how this handled. We shouldn’t just put the conditions of the old logical operator into the new one. I don’t think that’s how the user would expect an AND and OR logical operator to be combined.

diff_hunk: @@ -52,10 +83,48 @@ def apply_filter_policy(
           values from the runtime filters will overwrite those from the initial filters.
     :param init_filters: The initial filters set during the initialization of the relevant retriever.
     :param runtime_filters: The filters provided at runtime, usually during a query operation execution. These filters
-                            can change for each query/retreiver run invocation.
+                            can change for each query/retriever run invocation.
+    :param default_logical_operator: The default logical operator to use when merging filters (non-legacy filters only).
     :returns: A dictionary containing the resulting filters based on the provided policy.
     """
     if filter_policy == FilterPolicy.MERGE and runtime_filters:
-        return {**(init_filters or {}), **runtime_filters}
-    else:
-        return runtime_filters or init_filters
+        # legacy filters merge handling
+        if is_legacy(runtime_filters):
+            return {**(init_filters or {}), **runtime_filters}
+        elif init_filters is not None:
path: haystack/document_stores/types/filter_policy.py
text: @user1:
This is a bit confusing. Could we take out this if check to the beginning of this block and if init filters is None just return the runtime filters?