### Is there an existing issue for this?

- [X] I have searched the existing issues

### Feature description

AWS has a feature which allows you to [filter incoming lambda invocations using rules](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html). Without this feature I need to filter inside the lambda code itself which will incur more compute and cost in AWS. In the spirit of having as few diffs between localstack and AWS this is highly desireable for my pipeline.

(And fwiw, my pipeline is DynamoDB -> DynamoDB Stream -> Lambda (Python) -> Opensearch)

The current behavior is that create/update-event-source-mapping does parse (and validate) this option, but it isn't honored against incoming events at runtime.

### 🧑‍💻 Implementation

Implement with the same logic as AWS with support for all the operators in the linked article above.

### Anything else?

_No response_
return sqs_create_queue()
@pytest.fixture
def sqs_queue_arn(sqs_client):
    def _get_arn(queue_url: str) -> str:
        return sqs_client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["QueueArn"])[
            "Attributes"
        ]["QueueArn"]
    return _get_arn
@pytest.fixture
def sqs_queue_exists(sqs_client):
    def _queue_exists(queue_url: str) -> bool:
        """
        Checks whether a queue with the given queue URL exists.
        :param queue_url: the queue URL
        :return: true if the queue exists, false otherwise
        """
        try:
            result = sqs_client.get_queue_url(QueueName=queue_url.split("/")[-1])
            return result.get("QueueUrl") == queue_url
        except ClientError as e:
            if "NonExistentQueue" in e.response["Error"]["Code"]:
                return False
            raise
    yield _queue_exists
@pytest.fixture
def sns_create_topic(sns_client):
    topic_arns = []
    def _create_topic(**kwargs):