Support SQL filter expressions for streams

The JMS message selector parser should be re-usable in the future for
real JMS clients.

TODOs:
* Review
* Release notes
* Document both AMQP Filter Expressions and SQL Expressions in
  https://www.rabbitmq.com/docs/streams#filtering including how they can
  be used together with Bloom filters

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.
 # --------------------------------------------------------------------