Support SQL filter expressions for streams

 ## What?

This commit allows AMQP 1.0 clients to define SQL-like filter expressions when consuming from streams, enabling server-side message filtering.
RabbitMQ will only dispatch messages that match the provided filter expression, reducing network traffic and client-side processing overhead.
SQL filter expressions are a more powerful alternative to the [AMQP Property Filter Expressions](https://www.rabbitmq.com/blog/2024/12/13/amqp-filter-expressions) introduced in RabbitMQ 4.1.

SQL filter expressions are based on the [JMS message selector syntax](https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#message-selector-syntax) and support:
* Comparison operators (`=`, `<>`, `>`, `<`, `>=`, `<=`)
* Logical operators (`AND`, `OR`, `NOT`)
* Arithmetic operators (`+`, `-`, `*`, `/`)
* Special operators (`BETWEEN`, `LIKE`, `IN`, `IS NULL`)
* Access to the properties and application-properties sections

**Examples**

Simple expression:

```sql
header.priority > 4
```

Complex expression:

```sql
order_type IN ('premium', 'express') AND
total_amount BETWEEN 100 AND 5000 AND
(customer_region LIKE 'EU-%' OR customer_region = 'US-CA') AND
properties.creation-time >= 1750772279000 AND
NOT cancelled
```

Like AMQP property filter expressions, SQL filter expressions can be
combined with Bloom filters. Combining both allows for highly customisable
expressions (SQL) and extremely fast evaluation (Bloom filter) if only a
subset of the chunks need to be read from disk.

 ## Why?

Compared to AMQP property filter expressions, SQL filter expressions provide the following advantage:
* High expressiveness and flexibility in defining the filter

Like for AMQP property filter expressions, the following advantages apply:
* No false positives (as is the case for Bloom filters)
* Multiple concurrent clients can attach to the same stream each consuming
  only a specific subset of messages while preserving message order.
* Low network overhead as only messages that match the filter are
  transferred to the client
* Likewise, lower resource usage (CPU and memory) on clients since they
  don't need to deserialise messages that they are not interested in.
* If the SQL expression is simple, even the broker will save resources
  because it doesn't need to serialse and send messages that the client
  isn't interested in.

 ## How?

 ### JMS Message Selector Syntax vs. AMQP Extension Spec

The AMQP Filter Expressions Version 1.0 extension Working Draft 09 defines SQL Filter Expressions in Section 6.
This spec differs from the JMS message selector spec. Neither is a subset of the other. We can choose to follow either.
However, I think it makes most sense to follow the JMS spec because:
* The JMS spec is better defined
* The JMS spec is far more widespread than the AMQP Working Draft spec. (A slight variation of the AMQP Working
  Draft is used by Azure Service Bus: https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-messaging-sql-filter)
* The JMS spec is mostly simpler (partly because matching on only simple types)
* This will allow for a single SQL parser in RabbitMQ for both AMQP clients consuming from a stream and possibly in future for JMS clients consuming from queues or topics.

<details>
<summary>AMQP extension spec vs JMS spec</summary>

AMQP

!= is synonym for <>

JMS

defines only <>

Conclusion

<> is sufficient

AMQP

Strings can be tested for “greater than”

“both operands are of type string or of type symbol (any combination is permitted) and

the lexicographical rank of the left operand is greater than the lexicographical rank of the right operand”

JMS

“String and Boolean comparison is restricted to = and <>.”

Conclusion

The JMS behaviour is sufficient.

AMQP

IN <set-expression>

set-expression can contain non-string literals

JMS:

set-expression can contain only string literals

Conclusion

The JMS behaviour is sufficient.

AMQP

EXISTS predicate to check for composite types

JMS

Only simple types

Conclusion

We want to match only for simple types, i.e. allowing matching only against values in the application-properties, properties sections and priority field of the header section.

AMQP:

Modulo operator %

Conclusion

JMS doesn't define the modulo operator. Let's start without it.

We can decide in future to add support since it can actually be useful, for example for two receivers who want to process every other message.

AMQP:

The ‘+’ operator can concatenate string and symbol values

Conclusion

Such string concatenation isn't defined in JMS. We don't need it.

AMQP:

Define NAN and INF

JMS:

“Approximate literals use the Java floating-point literal syntax.”

Examples include "7."

Conclusion

We can go with the JMS spec given that needs to be implemented anyway
for JMS support.
Scientific notations are supported in both the AMQP spec and JMS spec.

AMQP

String literals can be surrounded by single or double quotation marks

JMS

A string literal is enclosed in single quotes

Conclusion

Supporting single quotes is good enough.

AMQP

“A binary constant is a string of pairs of hexadecimal digits prefixed by ‘0x’ that are not enclosed in quotation marks”

Conclusion

JMS doesn't support binary constants. We can start without binary constants.

Matching against binary values are still supported if these binary values can be expressed as UTF-8 strings.

AMQP

Functions DATE, UTC, SUBSTRING, LOWER, UPPER, LEFT, RIGHT

Vendor specific functions

Conclusion

JMS doesn't define such functions. We can start without those functions.

AMQP

<field><array_element_reference>

<field>‘.’<composite_type_reference>

to access map and array elements

Conclusion

Same as above:

We want to match only for simple types, i.e.  allowing matching only against values in the application-properties, properties sections and priority field of the header section.

AMQP

allows for delimited identifiers

JMS

Java identifier part characters

Conclusion

We can go with the Java identifiers extending the allowed characters by
`.` and `-` to reference field names such as `properties.group-id`.

JMS:

BETWEEN operator

Conclusion

The BETWEEN operator isn't supported in the AMQP spec. Let's support it as convenience since it's already available in JMS.

</details>

  ### Filter Name

The client provides a filter with name `sql-filter` instead of name
`jms-selector` to allow to differentiate between JMS clients and other
native AMQP 1.0 clients using SQL expressions. This way, we can also
optionally extend the SQL grammar in future.

  ### Identifiers

JMS message selectors allow identifiers to contain some well known JMS headers that match to well known AMQP fields, for example:
 ```erl
jms_header_to_amqp_field_name(<<"JMSDeliveryMode">>) -> durable;
jms_header_to_amqp_field_name(<<"JMSPriority">>) -> priority;
jms_header_to_amqp_field_name(<<"JMSMessageID">>) -> message_id;
jms_header_to_amqp_field_name(<<"JMSTimestamp">>) -> creation_time;
jms_header_to_amqp_field_name(<<"JMSCorrelationID">>) -> correlation_id;
jms_header_to_amqp_field_name(<<"JMSType">>) -> subject;
%% amqp-bindmap-jms-v1.0-wd10 § 3.2.2 JMS-defined ’JMSX’ Properties
jms_header_to_amqp_field_name(<<"JMSXUserID">>) -> user_id;
jms_header_to_amqp_field_name(<<"JMSXGroupID">>) -> group_id;
jms_header_to_amqp_field_name(<<"JMSXGroupSeq">>) -> group_sequence;
```

This commit does a similar matching for `header.` and `properties.` prefixed identifiers to field names in the AMQP property section.
The only field that is supported to filter on in the AMQP header section is `priority`, that is identifier `header.priority`.

By default, as described in the AMQP extension spec, if an identifier is not prefixed, it refers to a key in the application-properties section.

Hence, all identifiers prefixed with `header.`, and `properties.` have special meanings and MUST be avoided by applications unless they want to refer to those specific fields.

Azure Service Bus uses the `sys.` and `user.` prefixes for well known field names and arbitrary application-provided keys, respectively.

  ### SQL lexer, parser and evaluator

This commit implements the SQL lexer and parser in files rabbit_jms_selector_lexer.xrl and
rabbit_jms_selector_parser.yrl, respectively.

Advantages:
* Both the definitions in the lexer and the grammar in the parser are defined **declaratively**.
* In total, the entire SQL syntax and grammar is defined in only 240 lines.
* Therefore, lexer and parser are simple to maintain.

The idea of this commit is to use the same lexer and parser for native AMQP clients consumings
from streams (this commit) as for JMS clients (in the future).
All native AMQP client vs JMS client bits are then manipulated after
the Abstract Syntax Tree (AST) has been created by the parser.

For example, this commit transforms the AST specifically for native AMQP clients
by mapping `properties.` prefixed identifiers (field names) to atoms.
A JMS client's mapping from `JMS` prefixed headers can transform the AST
differently.

Likewise, this commit transforms the AST to compile a regex for complex LIKE
expressions when consuming from a stream while a future version
might not want to compile a regex when consuming from quorum queues.

Module `rabbit_jms_ast` provides such AST helper methods.

The lexer and parser are not performance critical as this work happens
upon receivers attaching to the stream.

The evaluator however is performance critical as message evaluation
happens on the hot path.

 ### LIKE expressions

The evaluator has been optimised to only compile a regex when necessary.
If the LIKE expression-value contains no wildcard or only a single `%`
wildcard, Erlang pattern matching is used as it's more efficient.
Since `_` can match any UTF-8 character, a regex will be compiled with
the `[unicode]` options.

  ### Filter errors

Any errors upon a receiver attaching to a stream causes the filter to
not become active. RabbitMQ will log a warning describing the reason and
will omit the named filter in its attach reply frame. The client lib is
responsible for detaching the link as explained in the AMQP spec:
> The receiving endpoint sets its desired filter, the sending endpoint sets the filter actually in place
(including any filters defaulted at the node). The receiving endpoint MUST check that the filter in
place meets its needs and take responsibility for detaching if it does not.

This applies to lexer and parser errors.

Errors during message evaluation will result in an unknown value.
Conditional operators on unknown are described in the JMS spec. If the
entire selector condition is unknown, the message does not match, and
will therefore not be delivered to the client.

 ## Clients

Support for passing the SQL expression from app to broker is provided by
the Java client in rabbitmq/rabbitmq-amqp-java-client#216

Author: ansd

deps/amqp10_client/include/amqp10_client.hrl

@@ -0,0 +1,11 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-record(filter, {
+          descriptor :: binary() | non_neg_integer(),
+          value :: term()
+         }).

deps/amqp10_client/src/amqp10_client.erl

@@ -7,7 +7,7 @@
 
 -module(amqp10_client).
 
--include("amqp10_client.hrl").
+-include("amqp10_client_internal.hrl").
 -include_lib("amqp10_common/include/amqp10_framing.hrl").
 
 -export([open_connection/1,

deps/amqp10_client/src/amqp10_client_connection.erl

@@ -9,7 +9,7 @@
 
 -behaviour(gen_statem).
 
--include("amqp10_client.hrl").
+-include("amqp10_client_internal.hrl").
 -include_lib("amqp10_common/include/amqp10_framing.hrl").
 -include_lib("amqp10_common/include/amqp10_types.hrl").
 

deps/amqp10_client/src/amqp10_client_frame_reader.erl

@@ -8,7 +8,7 @@
 
 -behaviour(gen_statem).
 
--include("amqp10_client.hrl").
+-include("amqp10_client_internal.hrl").
 -include_lib("amqp10_common/include/amqp10_framing.hrl").
 
 -ifdef(TEST).

deps/amqp10_client/src/amqp10_client.hrl renamed to deps/amqp10_client/src/amqp10_client_internal.hrl

@@ -9,6 +9,7 @@
 -behaviour(gen_statem).
 
 -include("amqp10_client.hrl").
+-include("amqp10_client_internal.hrl").
 -include_lib("amqp10_common/include/amqp10_framing.hrl").
 -include_lib("amqp10_common/include/amqp10_types.hrl").
 
@@ -86,7 +87,7 @@
 -type attach_role() :: {sender, target_def()} | {receiver, source_def(), pid()}.
 
 % http://www.amqp.org/specification/1.0/filters
--type filter() :: #{binary() => binary() | map() | list(binary())}.
+-type filter() :: #{binary() => #filter{} | binary() | map() | list(binary())}.
 -type max_message_size() :: undefined | non_neg_integer().
 -type footer_opt() :: crc32 | adler32.
 
@@ -781,29 +782,39 @@ translate_filters(Filters)
   when map_size(Filters) =:= 0 ->
     undefined;
 translate_filters(Filters) ->
-    {map,
-     maps:fold(
-       fun
-           (<<"apache.org:legacy-amqp-headers-binding:map">> = K, V, Acc) when is_map(V) ->
-               %% special case conversion
-               Key = sym(K),
-               [{Key, {described, Key, translate_legacy_amqp_headers_binding(V)}} | Acc];
-          (K, V, Acc) when is_binary(K) ->
-               %% try treat any filter value generically
-               Key = sym(K),
-               Value = filter_value_type(V),
-               [{Key, {described, Key, Value}} | Acc]
-       end, [], Filters)}.
-
-filter_value_type(V) when is_binary(V) ->
+    {map, lists:map(
+            fun({Name, #filter{descriptor = Desc,
+                               value = V}})
+                  when is_binary(Name) ->
+                    Descriptor = if is_binary(Desc) -> {symbol, Desc};
+                                    is_integer(Desc) -> {ulong, Desc}
+                                 end,
+                    {{symbol, Name}, {described, Descriptor, V}};
+               ({<<"apache.org:legacy-amqp-headers-binding:map">> = K, V})
+                 when is_map(V) ->
+                    %% special case conversion
+                    Key = sym(K),
+                    Val = translate_legacy_amqp_headers_binding(V),
+                    {Key, {described, Key, Val}};
+               ({K, V})
+                 when is_binary(K) ->
+                    Key = {symbol, K},
+                    Val = filter_value_type(V),
+                    {Key, {described, Key, Val}}
+            end, maps:to_list(Filters))}.
+
+filter_value_type(V)
+  when is_binary(V) ->
     %% this is clearly not always correct
     {utf8, V};
 filter_value_type(V)
   when is_integer(V) andalso V >= 0 ->
     {uint, V};
-filter_value_type(VList) when is_list(VList) ->
+filter_value_type(VList)
+  when is_list(VList) ->
     {list, [filter_value_type(V) || V <- VList]};
-filter_value_type({T, _} = V) when is_atom(T) ->
+filter_value_type({T, _} = V)
+  when is_atom(T) ->
     %% looks like an already tagged type, just pass it through
     V.
 
@@ -1507,16 +1518,17 @@ translate_filters_selector_filter_test() ->
     } = translate_filters(#{<<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}).
 
 translate_filters_multiple_filters_test() ->
-    {map,
-        [
-            {{symbol, <<"apache.org:selector-filter:string">>},
-             {described, {symbol, <<"apache.org:selector-filter:string">>},
-              {utf8, <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}}},
-            {{symbol, <<"apache.org:legacy-amqp-direct-binding:string">>},
-             {described, {symbol, <<"apache.org:legacy-amqp-direct-binding:string">>}, {utf8,<<"my topic">>}}}
-        ]
-    } = translate_filters(#{
-            <<"apache.org:legacy-amqp-direct-binding:string">> => <<"my topic">>,
-            <<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>
-        }).
+    {map, Actual} = translate_filters(
+                      #{
+                        <<"apache.org:legacy-amqp-direct-binding:string">> => <<"my topic">>,
+                        <<"apache.org:selector-filter:string">> => <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>
+                       }),
+    Expected = [{{symbol, <<"apache.org:selector-filter:string">>},
+                 {described, {symbol, <<"apache.org:selector-filter:string">>},
+                  {utf8, <<"amqp.annotation.x-opt-enqueuedtimeutc > 123456789">>}}},
+                {{symbol, <<"apache.org:legacy-amqp-direct-binding:string">>},
+                 {described, {symbol, <<"apache.org:legacy-amqp-direct-binding:string">>}, {utf8,<<"my topic">>}}}],
+    ActualSorted = lists:sort(Actual),
+    ExpectedSorted = lists:sort(Expected),
+    ExpectedSorted = ActualSorted.
 -endif.

deps/amqp10_client/src/amqp10_client_session.erl

@@ -16,7 +16,7 @@
          recv_amqp_header_step/1
         ]).
 
--include("src/amqp10_client.hrl").
+-include("src/amqp10_client_internal.hrl").
 
 start(Port) ->
     {ok, LSock} = gen_tcp:listen(Port, [binary, {packet, 0}, {active, false}]),

deps/amqp10_client/test/mock_server.erl

@@ -0,0 +1,31 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+
+%% A filter with this name contains a JMS message selector.
+%% We use the same name as sent by the Qpid JMS client in
+%% https://github.com/apache/qpid-jms/blob/2.7.0/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpSupport.java#L75
+-define(FILTER_NAME_JMS, <<"jms-selector">>).
+
+%% A filter with this name contains an SQL expression.
+%% In the current version, such a filter must comply with the JMS message selector syntax.
+%% However, we use a name other than "jms-selector" in case we want to extend the allowed syntax
+%% in the future, for example allowing for some of the extended grammar described in
+%% §6 "SQL Filter Expressions" of
+%% https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=66227
+-define(FILTER_NAME_SQL, <<"sql-filter">>).
+
+%% SQL-based filtering syntax
+%% These descriptors are defined in
+%% https://www.amqp.org/specification/1.0/filters
+-define(DESCRIPTOR_NAME_SELECTOR_FILTER, <<"apache.org:selector-filter:string">>).
+-define(DESCRIPTOR_CODE_SELECTOR_FILTER, 16#0000468C00000004).
+
+%% AMQP Filter Expressions Version 1.0 Working Draft 09
+%% https://groups.oasis-open.org/higherlogic/ws/public/document?document_id=66227
+-define(DESCRIPTOR_NAME_PROPERTIES_FILTER, <<"amqp:properties-filter">>).
+-define(DESCRIPTOR_CODE_PROPERTIES_FILTER, 16#173).
+-define(DESCRIPTOR_NAME_APPLICATION_PROPERTIES_FILTER, <<"amqp:application-properties-filter">>).
+-define(DESCRIPTOR_CODE_APPLICATION_PROPERTIES_FILTER, 16#174).

deps/amqp10_common/include/amqp10_filter.hrl

@@ -258,7 +258,7 @@ define ct_master.erl
 endef
 
 PARALLEL_CT_SET_1_A = unit_rabbit_ssl unit_cluster_formation_locking_mocks unit_cluster_formation_sort_nodes unit_collections unit_config_value_encryption unit_connection_tracking
-PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_credit_api_v2 amqp_filtex amqp_dotnet amqp_jms signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management
+PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_credit_api_v2 amqp_filter_prop amqp_filter_sql amqp_jms_unit amqp_dotnet amqp_jms signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management
 PARALLEL_CT_SET_1_C = amqp_proxy_protocol amqpl_consumer_ack amqpl_direct_reply_to backing_queue bindings rabbit_db_maintenance rabbit_db_msup rabbit_db_policy rabbit_db_queue rabbit_db_topic_exchange rabbit_direct_reply_to_prop cluster_limit cluster_minority term_to_binary_compat_prop topic_permission transactions unicode unit_access_control
 PARALLEL_CT_SET_1_D = amqqueue_backward_compatibility channel_interceptor channel_operation_timeout classic_queue classic_queue_prop config_schema peer_discovery_dns peer_discovery_tmp_hidden_node per_node_limit per_user_connection_channel_limit
 
@@ -363,6 +363,9 @@ ifdef TRACE_SUPERVISOR2
 RMQ_ERLC_OPTS += -DTRACE_SUPERVISOR2=true
 endif
 
+# https://www.erlang.org/doc/apps/parsetools/leex.html#file/2
+export ERL_COMPILER_OPTIONS := deterministic
+
 # --------------------------------------------------------------------
 # Documentation.
 # --------------------------------------------------------------------

deps/amqp10_common/include/amqp10_filtex.hrl

@@ -45,7 +45,7 @@
 -type str() :: atom() | string() | binary().
 -type internal_ann_key() :: atom().
 -type x_ann_key() :: binary(). %% should begin with x- or ideally x-opt-
--type x_ann_value() :: str() | integer() | float() | TaggedValue :: tuple() | [x_ann_value()].
+-type x_ann_value() :: str() | number() | TaggedValue :: tuple() | [x_ann_value()].
 -type protocol() :: module().
 -type annotations() :: #{internal_ann_key() => term(),
                          x_ann_key() => x_ann_value()}.
@@ -76,8 +76,7 @@
 -type property_value() :: undefined |
                           string() |
                           binary() |
-                          integer() |
-                          float() |
+                          number() |
                           boolean().
 -type tagged_value() :: {uuid, binary()} |
                         {utf8, binary()} |
@@ -155,9 +154,9 @@ init(Proto, Data, Anns) ->
 -spec init(protocol(), term(), annotations(), environment()) -> state().
 init(Proto, Data, Anns0, Env) ->
     {ProtoData, ProtoAnns} = Proto:init(Data),
-    Anns1 = case map_size(Env) =:= 0 of
-                true -> Anns0;
-                false -> Anns0#{env => Env}
+    Anns1 = case map_size(Env) of
+                0 -> Anns0;
+                _ -> Anns0#{env => Env}
             end,
     Anns2 = maps:merge(ProtoAnns, Anns1),
     Anns = ensure_received_at_timestamp(Anns2),

deps/rabbit/Makefile

@@ -0,0 +1,24 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries.  All rights reserved.
+
+-module(rabbit_amqp_filter).
+
+-export([eval/2]).
+
+-type expression() :: undefined |
+                      {property, rabbit_amqp_filter_prop:parsed_expressions()} |
+                      {jms, rabbit_amqp_filter_jms:parsed_expression()}.
+
+-export_type([expression/0]).
+
+-spec eval(expression(), mc:state()) -> boolean().
+eval(undefined, _Mc) ->
+    %% A receiver without filter wants all messages.
+    true;
+eval({property, Expr}, Mc) ->
+    rabbit_amqp_filter_prop:eval(Expr, Mc);
+eval({jms, Expr}, Mc) ->
+    rabbit_amqp_filter_jms:eval(Expr, Mc).