Support AMQP SQL Filter Expressions

Instead of the JMS message selector syntax, support a subset of the
AMQP SQL Filter Expressions synatx as defined in
AMQP Filter Expressions Version 1.0 Committee Specification Draft 01
https://docs.oasis-open.org/amqp/filtex/v1.0/csd01/filtex-v1.0-csd01.html#_Toc67929276

This commit changes the descriptors.

Author: ansd

deps/amqp10_common/include/amqp10_filter.hrl

@@ -4,28 +4,15 @@
 %%
 %% 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
+%% AMQP Filter Expressions Version 1.0 Committee Specification Draft 01
+%% https://docs.oasis-open.org/amqp/filtex/v1.0/csd01/filtex-v1.0-csd01.html#_Toc67929266
 -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).
+
+%% A filter with this name contains an AMQP SQL expression.
+-define(FILTER_NAME_SQL, <<"sql-filter">>).
+%% https://docs.oasis-open.org/amqp/filtex/v1.0/csd01/filtex-v1.0-csd01.html#_Toc67929276
+-define(DESCRIPTOR_NAME_SQL_FILTER, <<"amqp:sql-filter">>).
+-define(DESCRIPTOR_CODE_SQL_FILTER, 16#120).

deps/rabbit/src/rabbit_amqp_filter_jms.erl

@@ -291,9 +291,9 @@ get_field_value(Name, Msg) ->
             undefined
     end.
 
-check_descriptor({symbol, ?DESCRIPTOR_NAME_SELECTOR_FILTER}) ->
+check_descriptor({symbol, ?DESCRIPTOR_NAME_SQL_FILTER}) ->
     ok;
-check_descriptor({ulong, ?DESCRIPTOR_CODE_SELECTOR_FILTER}) ->
+check_descriptor({ulong, ?DESCRIPTOR_CODE_SQL_FILTER}) ->
     ok;
 check_descriptor(_) ->
     error.

deps/rabbit/test/amqp_filter_sql_SUITE.erl

@@ -338,7 +338,7 @@ invalid_filter(Config) ->
                   #{arguments => #{<<"x-queue-type">> => {utf8, <<"stream">>}}}),
 
     %% Trigger a lexer error.
-    Filter1 = #{?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_CODE_SELECTOR_FILTER,
+    Filter1 = #{?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_CODE_SQL_FILTER,
                                             value = {utf8, <<"@#$%^&">>}}},
     {ok, Receiver1} = amqp10_client:attach_receiver_link(
                         Session, <<"receiver 1">>, Address,
@@ -356,7 +356,7 @@ invalid_filter(Config) ->
     ok = detach_link_sync(Receiver1),
 
     %% Trigger a parser error. We use allowed tokens here, but the grammar is incorrect.
-    Filter2 = #{?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_CODE_SELECTOR_FILTER,
+    Filter2 = #{?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_CODE_SQL_FILTER,
                                             value = {utf8, <<"FALSE FALSE">>}}},
     {ok, Receiver2} = amqp10_client:attach_receiver_link(
                         Session, <<"receiver 2">>, Address,
@@ -375,7 +375,7 @@ invalid_filter(Config) ->
     PropsFilter = [{{symbol, <<"subject">>}, {utf8, <<"some subject">>}}],
     Filter3 = #{<<"prop name">> => #filter{descriptor = ?DESCRIPTOR_NAME_PROPERTIES_FILTER,
                                            value = {map, PropsFilter}},
-                ?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_CODE_SELECTOR_FILTER,
+                ?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_CODE_SQL_FILTER,
                                             value = {utf8, <<"TRUE">>}}},
     {ok, Receiver3} = amqp10_client:attach_receiver_link(
                         Session, <<"receiver 3">>, Address,
@@ -393,7 +393,7 @@ invalid_filter(Config) ->
 
     %% Send invalid UTF-8 in the SQL expression.
     InvalidUTF8 = <<255>>,
-    Filter4 = #{?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_CODE_SELECTOR_FILTER,
+    Filter4 = #{?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_CODE_SQL_FILTER,
                                             value = {utf8, InvalidUTF8}}},
     {ok, Receiver4} = amqp10_client:attach_receiver_link(
                         Session, <<"receiver 4">>, Address,
@@ -432,7 +432,7 @@ filter(String)
   when is_binary(String) ->
     #{<<"from start">> => #filter{descriptor = <<"rabbitmq:stream-offset-spec">>,
                                   value = {symbol, <<"first">>}},
-      ?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_NAME_SELECTOR_FILTER,
+      ?FILTER_NAME_SQL => #filter{descriptor = ?DESCRIPTOR_NAME_SQL_FILTER,
                                   value = {utf8, String}}}.
 
 assert_credit_exhausted(Receiver, Line) ->

deps/rabbit/test/amqp_jms_unit_SUITE.erl

@@ -884,7 +884,7 @@ match(Selector, Header, Props, AppProps)
     rabbit_amqp_filter_jms:eval(ParsedSelector, Mc).
 
 parse(Selector) ->
-    Descriptor = {ulong, ?DESCRIPTOR_CODE_SELECTOR_FILTER},
+    Descriptor = {ulong, ?DESCRIPTOR_CODE_SQL_FILTER},
     Filter = {described, Descriptor, {utf8, Selector}},
     rabbit_amqp_filter_jms:parse(Filter).