[ROCKETMQ-121]Support message filtering based on SQL92 #82

vsair · 2017-03-22T07:23:08Z

Support message filtering based on SQL92

So far, RocketMQ only support message filtering feature by TAG, but one message only can own one tag, this is too limited to meet complex business requirements.

So, we want to define and implement a reasonable filter language based on a subset of the SQL 92 expression syntax to support customized message filtering.

Why subset of SQL92

Let RocketMQ has the ability of message filtering is the purpose of this issue, no matter SQL92 or any other languages.

As I know, ActiveMQ already impllement this functionality based on JavaCC, it's simple and exntensible.So I just extract it and integrate into RocketMQ, only some grammars:

numeric comparison, like >, >=, <,<=, BETWEEN
character comparison, like =, <>, IN
check NULL
logical AND, logical OR, logical NOT

Design

New Module, rocketmq-filter

The implementation of SQL92 language is placed in this module which have dependency on common module.

Broker compile or evaluate expression through the interface of FilterSpi contained in FilterFactory that manage all FilterSpi and also support new one to register.

How to manage consumer's expression data

Different from tag filtering, expression of SQL92 should be compiled first to check whether is leagal and then use the complied expression to compute. This procedure is designed to take place at broker.

ConsumerManager manage the suscriptions of push consumer, and ConsumerFilterManager manage the expression info of push consumer who wish to filter message by special language, the info includes data version, expression, compiled expression, alive time and etc.

How to filter message by expression

I redesign the interface getMessage of MessageStore by replace the last parameter SubscriptionData to MessageFilter that is also refactored. The purpose is to make module rocketmq-store has no relation with protocol.

When get message, the implementation ExpressionMessageFilter would check whether the message is matched by BitsArray which will be refered later or evaluation, just as the mechanism of tag filtering.

Optimization, pre-calculate the filtering result when build consume queue

It's poor performance to do filter when pull message:

off-heap to heap, once every consumer subscribed same topic pull message.
decode message properties, once every consumer subscribed same topic pull message.

BloomFilter and pre-calculation are adopted to optimize the situation:

Every consumer has been asigned some bit position of BloomFilter when register to broker.
When broker build queue after message into CommitLog, the consumer's filtering result would be calculated, and all resuls are assembled as a BitsArray saved in ConsumeQueueExt.
ConsumeQueueExt is a store file linked to ConsumeQueue, ConsumeQueue could find the data by the tagsCode whitch is already replaced by the address(for compitable, the range is Long.MIN_VALUE to Integer.MIN_VALUE) generated by ConsumeQueueExt.
ExpressionMessageFilter could use the BitsArray to check whether the message is matched. Because of BloomFilter's collision, it also need to decode properties to do calculation for matched message(may could be reduced by check the collision, not include in this edition).

This optimization is suitable for:

High subscription ratio.
Large properties.

Interface

Only push consumer could filter message by SQL92 expression in this edition.

coveralls · 2017-03-22T07:41:15Z

Coverage increased (+3.4%) to 35.261% when pulling e258309 on vsair:ROCKETMQ-121 into 72e6def on apache:develop.

coveralls · 2017-03-22T07:41:15Z

Coverage increased (+3.4%) to 35.261% when pulling e258309 on vsair:ROCKETMQ-121 into 72e6def on apache:develop.

coveralls · 2017-03-23T03:14:52Z

Coverage increased (+3.8%) to 35.617% when pulling a9cea30 on vsair:ROCKETMQ-121 into 72e6def on apache:develop.

coveralls · 2017-03-23T03:14:52Z

Coverage increased (+3.8%) to 35.617% when pulling a9cea30 on vsair:ROCKETMQ-121 into 72e6def on apache:develop.

coveralls · 2017-03-23T03:14:52Z

Coverage increased (+3.8%) to 35.617% when pulling a9cea30 on vsair:ROCKETMQ-121 into 72e6def on apache:develop.

vongosling · 2017-03-30T06:29:07Z

@shroman @lizhanhui @lollipopjin Could you help us to review this great job ?

vongosling · 2017-03-30T09:38:23Z

broker/src/main/java/org/apache/rocketmq/broker/client/ConsumerIdsChangeListener.java

 public interface ConsumerIdsChangeListener {
-    void consumerIdsChanged(final String group, final List<Channel> channels);
+


Great refactoring

coveralls · 2017-03-30T10:29:08Z

Coverage increased (+3.7%) to 35.609% when pulling 67fe978 on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

coveralls · 2017-03-30T10:29:08Z

Coverage increased (+3.7%) to 35.609% when pulling 67fe978 on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

coveralls · 2017-03-30T10:29:08Z

Coverage increased (+3.7%) to 35.609% when pulling 67fe978 on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

vongosling · 2017-03-30T09:44:06Z

broker/src/main/java/org/apache/rocketmq/broker/filter/CommitLogDispatcherCalcBitMap.java

+
+import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.filter.util.BitsArray;


BitMap ? Could we use the roaring bitmap, https://github.com/RoaringBitmap/RoaringBitmap

vongosling · 2017-03-30T09:45:37Z

broker/src/main/java/org/apache/rocketmq/broker/filter/CommitLogDispatcherCalcBitMap.java

+
+                    ret = filterData.getCompiledExpression().evaluate(context);
+                } catch (Throwable e) {
+                    log.error("Calc filter bit map error!commitLogOffset=" + request.getCommitLogOffset() +


Please use {} replace +

vongosling · 2017-03-30T09:46:50Z

broker/src/main/java/org/apache/rocketmq/broker/filter/CommitLogDispatcherCalcBitMap.java

+                log.warn("Spend {} ms to calc bit map, consumerNum={}, topic={}", eclipseTime, filterDatas.size(), request.getTopic());
+            }
+        } catch (Throwable e) {
+            log.error("Calc bit map error! topic=" + request.getTopic() + ", offset=" + request.getCommitLogOffset()


Follow the previous comment

vongosling · 2017-03-30T09:47:50Z

broker/src/main/java/org/apache/rocketmq/broker/filter/ConsumerFilterData.java

+        return this.deadTime >= this.bornTime;
+    }
+
+    public long deadHowLong() {


what is deadhowlong

vongosling · 2017-03-30T09:49:38Z

broker/src/main/java/org/apache/rocketmq/broker/filter/ConsumerFilterData.java

+    }
+
+    @Override
+    public boolean equals(Object o) {


Could we use ToStringBuilder in Commons Lang3 to build equal and hash override method?

Already replaced by EqualsBuilder, ToStringBuilder and HashCodeBuilder.

vongosling · 2017-03-30T09:59:08Z

filter/src/main/java/org/apache/rocketmq/filter/expression/ConstantExpression.java

+ * Represents a constant expression
+ * <p>
+ * This class was taken from ActiveMQ org.apache.activemq.filter.ConstantExpression,
+ * but:


vongosling · 2017-03-30T10:00:34Z

filter/src/main/java/org/apache/rocketmq/filter/parser/ParseException.java

+ * limitations under the License.
+ */
+
+/* Generated By:JavaCC: Do not edit this line. ParseException.java Version 5.0 */


why java Version 5.0

vongosling · 2017-03-30T10:04:57Z

filter/src/main/java/org/apache/rocketmq/filter/util/HashAlgorithm.java

+ * significant. This allows a positive 32-bit number to be returned for all
+ * cases.
+ * <br>Don't change the order of algorithms, add new algorithm to last if you want.
+ */


Recommended commons codec's hashing algorithms or guava hash

Ok, I'll check whether the third party's jar could meet the scene.

vongosling · 2017-03-30T10:06:04Z

filter/src/main/java/org/apache/rocketmq/filter/util/BloomFilter.java

+/**
+ * Simple implement of bloom filter.
+ */
+public class BloomFilter {


Recommended existent mature BloomFilter in guava or hahoop‘s

Hi, I replaced HashAlgorithm to Guava Hashing.murmur3_128. Most classes of Guava's BloomFilter are private and final, which could not be included by RocketMQ, such as BitArray saved in store.So I didn't use it.

Great, murmur3_128

vongosling · 2017-03-30T10:07:20Z

filter/src/main/java/org/apache/rocketmq/filter/util/LRUCache.java

+ * @param <K>
+ * @param <V>
+ */
+public class LRUCache<K, V> extends LinkedHashMap<K, V> {


Can we replace with guava CacheBuilder?

coveralls · 2017-04-11T03:23:04Z

Coverage increased (+3.7%) to 35.602% when pulling 0692568 on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

coveralls · 2017-04-11T03:23:04Z

Coverage increased (+3.7%) to 35.602% when pulling 0692568 on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

coveralls · 2017-04-11T03:23:04Z

Coverage increased (+3.7%) to 35.602% when pulling 0692568 on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

coveralls · 2017-04-11T08:55:40Z

Coverage increased (+3.4%) to 35.262% when pulling 24f6ada on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

coveralls · 2017-04-11T08:55:40Z

Coverage increased (+3.4%) to 35.262% when pulling 24f6ada on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

coveralls · 2017-04-11T08:55:40Z

Coverage increased (+3.4%) to 35.262% when pulling 24f6ada on vsair:ROCKETMQ-121 into 45a64fd on apache:develop.

lizhanhui · 2017-04-14T02:32:50Z

Will look into this PR today.

vongosling · 2017-04-17T11:45:03Z

I think only servtool‘s module has guava dependency, so there is no transitive dependency pollution on sdk, right?

vongosling · 2017-04-17T11:51:55Z

please @lizhanhui @shroman help to review this great PR :-)

vsair · 2017-04-17T13:05:22Z

@vongosling yeah, only server's modules have guava dependency, client doesn't.

zhouxinyu · 2017-04-20T11:58:51Z

@lizhanhui , Do you have any other thoughts about this PR? If no, I will merge this PR soon.

And please @vsair help resolve the conflicting files.

lizhanhui · 2017-04-20T23:47:01Z

@zhouxinyu I have checked the major data flow and it looks good to me. I have not scrutinized the changes line by line yet. You may merge it first and we may discuss potential issues hereafter.

1. Add filter module 2. Manage consumer filter expression 3. Support pre calculate filter result when build consume queue. 4. Check whether server support feature of sql when consumer start, maybe it's not the best solution (When network is not stable?).

1. Srvutil module include Guava 2. Replace LRUCache to CacheBuilder, replace HashAlgorithm to Guava Hashing.murmur3_128

coveralls · 2017-04-21T02:18:53Z

Coverage increased (+3.3%) to 37.892% when pulling 09dca78 on vsair:ROCKETMQ-121 into 42f78c2 on apache:develop.

coveralls · 2017-04-21T02:18:53Z

Coverage increased (+3.3%) to 37.892% when pulling 09dca78 on vsair:ROCKETMQ-121 into 42f78c2 on apache:develop.

coveralls · 2017-04-21T02:18:53Z

Coverage increased (+3.3%) to 37.892% when pulling 09dca78 on vsair:ROCKETMQ-121 into 42f78c2 on apache:develop.

vsair · 2017-04-21T02:28:55Z

@zhouxinyu Conflicts have been resolved. Thanks.

dongeforever · 2017-04-21T10:13:19Z

It seems that this PR could be merged now. @vongosling @zhouxinyu @lizhanhui

Jaskey · 2017-06-12T06:47:51Z

@vsair

I have two things to confirms,

if the old messages which exists before consumer subscribe with the filter expression must be filtered by decodng properties to do calculation ?
expectConsumerNumUseFilter = 32 and maxErrorRateOfBloomFilter = 20 can be configured and changed, what if I change this value and restart, does the calculated results in the ConsumeQueueExt updated accordingly?

vsair · 2017-06-12T07:57:00Z

Hi, Jaskey; Thanks for your question. 1. It will decode the properties to do calculation at that case. 2. When they are changed, the consumer's subscription will be ignored when starting, so the results int bloom filter also will be ignored. That is mean server will do calculation again and not update the results already existed in Store. Thanks, Vsair

…

On Mon, Jun 12, 2017 at 2:47 PM, Jaskey ***@***.***> wrote: @vsair <https://github.com/vsair> I have two things to confirms, 1. if the old messages which exists before consumer subscribe with the filter expression must be filtered by decodng properties to do calculation ? 2. `expectConsumerNumUseFilter = 32` and `maxErrorRateOfBloomFilter = 20` can be configured and changed, what if I change this value and restart, does the calculated results in the ConsumeQueueExt updated accordingly? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#82 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ACeujdZf1ZNbu-MycGfHW_kbIuCCvUE3ks5sDN8ZgaJpZM4Mk0bH> .

Jaskey · 2017-06-12T09:25:26Z

@vsair

Thanks for your clarification. I am just trying to go through the main logic of sql filter.

For the point of

When they are changed, the consumer's subscription will be ignored when
starting, so the results int bloom filter also will be ignored. That is
mean server will do calculation again and not update the results already
existed in Store.

where is that code logic ? I search for usage of getExpectConsumerNumUseFilter but I find that it's only used to create a new bloom filter instance when starting , then how does broker detect that the configuration is changed? Are we saying the same thing.

What I am concerned is that, if we change any configuration about the bloom filter, the bit result may be different as it was before changing configuration , so it will be different from the result persisted in the consume queue ext. I wonder that will broker works as usual to filter the expected messages as it did before.

improving codeCov for #22

vsair force-pushed the ROCKETMQ-121 branch from e258309 to a9cea30 Compare March 23, 2017 02:58

vsair changed the title ~~[Rocketmq 121]Support message filtering based on SQL92~~ [Rocketmq-121]Support message filtering based on SQL92 Mar 27, 2017

vsair changed the title ~~[Rocketmq-121]Support message filtering based on SQL92~~ [ROCKETMQ-121]Support message filtering based on SQL92 Mar 27, 2017

vongosling reviewed Mar 30, 2017

View reviewed changes

vsair force-pushed the ROCKETMQ-121 branch from a9cea30 to 67fe978 Compare March 30, 2017 10:13

vongosling reviewed Mar 30, 2017

View reviewed changes

vsair force-pushed the ROCKETMQ-121 branch from 67fe978 to 0692568 Compare April 11, 2017 03:07

vsair added 2 commits April 21, 2017 09:54

[ROCKETMQ-121] RocketMQ support filer message by SQL

fa97c79

1. Srvutil module include Guava 2. Replace LRUCache to CacheBuilder, replace HashAlgorithm to Guava Hashing.murmur3_128

[ROCKETMQ-121] Fix conflicts of MixAll util.

09dca78

vsair force-pushed the ROCKETMQ-121 branch from 24f6ada to 09dca78 Compare April 21, 2017 02:02

asfgit pushed a commit that referenced this pull request Apr 21, 2017

[ROCKETMQ-121]Support message filtering based on SQL92 closes #82

58f1574

vsair closed this Apr 24, 2017

vsair mentioned this pull request Apr 26, 2017

[ROCKETMQ-121]Docs of filtering messages based on SQL92 apache/rocketmq-site#11

Closed

asfgit pushed a commit that referenced this pull request Jun 6, 2017

[ROCKETMQ-121]Support message filtering based on SQL92 closes #82

9eeb2f7

JiaMingLiu93 pushed a commit to JiaMingLiu93/rocketmq that referenced this pull request May 28, 2020

[ROCKETMQ-121]Support message filtering based on SQL92 closes apache#82

246b3cc

JiaMingLiu93 pushed a commit to JiaMingLiu93/rocketmq that referenced this pull request May 28, 2020

[ROCKETMQ-121]Support message filtering based on SQL92 closes apache#82

633b372

pingww pushed a commit that referenced this pull request Aug 26, 2022

Merge pull request #82 from YxAc/continuously_improve_codeCov

011fbb3

improving codeCov for #22

		public interface ConsumerIdsChangeListener {
		void consumerIdsChanged(final String group, final List<Channel> channels);

[ROCKETMQ-121]Support message filtering based on SQL92 #82

[ROCKETMQ-121]Support message filtering based on SQL92 #82

Conversation

vsair commented Mar 22, 2017

Support message filtering based on SQL92

Why subset of SQL92

Design

Interface

coveralls commented Mar 22, 2017

coveralls commented Mar 22, 2017

coveralls commented Mar 23, 2017

coveralls commented Mar 23, 2017

coveralls commented Mar 23, 2017

vongosling commented Mar 30, 2017 • edited

Choose a reason for hiding this comment

coveralls commented Mar 30, 2017

coveralls commented Mar 30, 2017

coveralls commented Mar 30, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vongosling Mar 30, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vongosling Mar 30, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vongosling Mar 30, 2017 • edited

Choose a reason for hiding this comment

coveralls commented Apr 11, 2017

coveralls commented Apr 11, 2017

coveralls commented Apr 11, 2017

coveralls commented Apr 11, 2017

coveralls commented Apr 11, 2017

coveralls commented Apr 11, 2017

lizhanhui commented Apr 14, 2017

vongosling commented Apr 17, 2017 • edited

vongosling commented Apr 17, 2017

vsair commented Apr 17, 2017

zhouxinyu commented Apr 20, 2017

lizhanhui commented Apr 20, 2017

coveralls commented Apr 21, 2017

coveralls commented Apr 21, 2017

coveralls commented Apr 21, 2017

vsair commented Apr 21, 2017

dongeforever commented Apr 21, 2017

Jaskey commented Jun 12, 2017 • edited

vsair commented Jun 12, 2017 via email

Jaskey commented Jun 12, 2017 • edited

vongosling commented Mar 30, 2017 •

edited

vongosling Mar 30, 2017 •

edited

vongosling Mar 30, 2017 •

edited

vongosling Mar 30, 2017 •

edited

vongosling commented Apr 17, 2017 •

edited

Jaskey commented Jun 12, 2017 •

edited

Jaskey commented Jun 12, 2017 •

edited