Repository navigation
[ISSUE #7050]🚀feat(filter): enhance SQL92 expression support with new operators and improve evaluation context - #7051
Merged
Merged
Conversation
… operators and improve evaluation context
rocketmq-rust-bot
requested review from
SpaceXCN,
TeslaRustor and
rocketmq-rust-bot
April 4, 2026 12:42
Collaborator
|
🔊@mxsm 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
rocketmq-rust-bot
added
AI review first
Contributor
WalkthroughThis PR enhances SQL92 expression filtering by introducing new operators (NOW, IN/NOT IN, BETWEEN, CONTAINS, STARTSWITH, ENDSWITH), refactoring the evaluation context from Changes
Sequence Diagram(s)sequenceDiagram
actor Consumer
participant Processor as Pop/Pull Processor
participant Manager as ConsumerFilterManager
participant Cache as Expression Cache
participant Compiler as SQL Compiler
participant FilterData as ConsumerFilterData
participant Evaluator as ExpressionMessageFilter
Consumer->>Processor: Request (expression, type, version)
Processor->>Manager: resolve(topic, group, expr, type, version)
Manager->>Cache: lookup(type, expr)
alt Cache Hit - Compiled
Cache-->>Manager: Arc<Expression>
Manager->>FilterData: set_compiled_expression_arc(cached)
Manager-->>Processor: Some(ConsumerFilterData)
else Cache Miss - Needs Compilation
Manager->>Cache: lookup_failure(type, expr)
alt Previously Failed
Cache-->>Manager: None (negative cache)
Manager-->>Processor: None (resolution failed)
else First Attempt
Manager->>Compiler: compile(expr)
alt Compilation Success
Compiler-->>Manager: Expression
Manager->>Cache: store(type, expr, Arc<Expression>)
Manager->>FilterData: set_compiled_expression_arc(compiled)
Manager-->>Processor: Some(ConsumerFilterData)
else Compilation Failure
Compiler-->>Manager: Error
Manager->>Cache: store_failure(type, expr)
Manager-->>Processor: None (resolution failed)
end
end
end
Processor->>Evaluator: match_message(msg, filter_data)
Evaluator->>Evaluator: build MessageEvaluationContext(properties)
Evaluator->>Compiler: evaluate(expression, context)
Compiler-->>Evaluator: boolean result
Evaluator-->>Processor: match result
Estimated code review effort🎯 4 (Complex) | ⏱️ ~70 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (2)
rocketmq-broker/src/filter/manager/consumer_filter_manager.rs (2)
312-338: Correct thread-safe caching with Entry API, but contains unreachable code.The
compile_with_cacheimplementation correctly uses DashMap's Entry API for thread-safe cache population. However, line 333 (self.failed_expression_cache.remove(entry.key())) is unreachable code - if the key were infailed_expression_cache, we would have returnedNoneat line 314-316 before reaching this point.🧹 Remove unreachable code
Err(error) => { self.failed_expression_cache .insert(entry.key().clone(), error.to_string()); return None; } }; - self.failed_expression_cache.remove(entry.key()); entry.insert(compiled.clone()); Some(compiled)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rocketmq-broker/src/filter/manager/consumer_filter_manager.rs` around lines 312 - 338, In compile_with_cache, remove the unreachable call to self.failed_expression_cache.remove(entry.key()) (in the Entry::Vacant branch) because the method already returns early if failed_expression_cache contains the key; update the function by deleting that line so there is no dead code; the rest of the logic in compile_with_cache, FilterFactory::instance().get, filter.compile, and the compiled_expression_cache.entry handling remains unchanged.
48-49: Consider adding cache size limits or TTL to prevent unbounded memory growth.The
compiled_expression_cacheandfailed_expression_cachegrow without bounds and are only cleared onstop(). For long-running brokers with many unique filter expressions, this could lead to memory pressure.Consider adding:
- A maximum cache size with LRU eviction, or
- A periodic cleanup based on access time, or
- At minimum, documenting the expected cardinality of unique expressions
This is likely acceptable for typical deployments where the number of unique SQL expressions is limited, but worth considering for large-scale scenarios.
Also applies to: 67-68
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rocketmq-broker/src/filter/manager/consumer_filter_manager.rs` around lines 48 - 49, The compiled_expression_cache and failed_expression_cache in ConsumerFilterManager are unbounded; change their implementation to a bounded/TTL cache to avoid unbounded memory growth — replace or wrap the Arc<DashMap<String, ...>> fields (compiled_expression_cache, failed_expression_cache) with a thread-safe LRU or time-aware cache (e.g., moka, lru_time_cache, or an LRU wrapper around DashMap) and add eviction policies in the ConsumerFilterManager constructor/new path, ensure cache lookups/inserts in methods that currently access compiled_expression_cache and failed_expression_cache follow the new API, and add a background cleanup task or TTL metadata if using time-based eviction; keep stop() behavior but remove reliance on it for memory reclamation.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rocketmq-filter/src/filter/sql_runtime.rs`:
- Around line 137-152: The IN/NOT IN branch (ExprNode::InList) currently ignores
unknown (NULL) comparison results from eval_compare_values and therefore
collapses expressions like x IN (1, NULL) incorrectly; modify the loop to track
an "unknown" flag when eval_compare_values returns None, set matched=true if
Some(true), and after the loop return Value::Null if no definite match and
unknown is true; otherwise return the boolean (applying negation when *negated).
Reference ExprNode::InList and eval_compare_values (and alternatively
parse_constant_value if you prefer rejecting NULL members at parse time).
- Line 109: ExprNode::Now currently calls current_millis() per node, so make NOW
stable by capturing a single timestamp at the start of an evaluation and reusing
it for every ExprNode::Now in that invocation: modify SqlExpression::evaluate()
(or the evaluation context passed into node evaluation) to compute let now_ts =
current_millis() once and then change ExprNode::Now handling to return
Value::Long(now_ts as i64) (or read now_ts from the shared context) so all NOW
nodes in one evaluate() use the same value.
- Around line 192-206: The early return on search.is_empty() bypasses negation;
change the logic so you compute the base match first and then apply negation:
instead of returning Ok(Value::Boolean(false)) when search.is_empty(), set
matched = true (or otherwise compute the same base result you would for an empty
pattern) and then return Ok(Value::Boolean(if *negated { !matched } else {
matched })); adjust the block around expr.evaluate(context) /
StringMatchOp::{Contains,StartsWith,EndsWith} to handle empty search as a
regular match case (or alternatively enforce non-empty RHS at parse time).
- Around line 279-299: The numeric comparison currently uses coerce_numeric to
convert everything to f64 (in eval_compare_values), causing precision loss for
large i64 values; change the logic in eval_compare_values so that if both
operands are integral (Value::Long or integer-string) you compare them as i64
directly (using i64 partial_cmp / equality) and only fall back to f64
comparisons when at least one operand is a floating-point value; update or add
helper coercion behavior (coerce_numeric or a new coerce_to_numeric_enum) so it
returns a discriminated numeric kind (e.g., Int(i64) vs Float(f64)) and use that
to call numeric_eq or ordering appropriately, keeping existing EvaluationError
paths (and using numeric_eq for float branch) and preserving the existing
early-null handling.
---
Nitpick comments:
In `@rocketmq-broker/src/filter/manager/consumer_filter_manager.rs`:
- Around line 312-338: In compile_with_cache, remove the unreachable call to
self.failed_expression_cache.remove(entry.key()) (in the Entry::Vacant branch)
because the method already returns early if failed_expression_cache contains the
key; update the function by deleting that line so there is no dead code; the
rest of the logic in compile_with_cache, FilterFactory::instance().get,
filter.compile, and the compiled_expression_cache.entry handling remains
unchanged.
- Around line 48-49: The compiled_expression_cache and failed_expression_cache
in ConsumerFilterManager are unbounded; change their implementation to a
bounded/TTL cache to avoid unbounded memory growth — replace or wrap the
Arc<DashMap<String, ...>> fields (compiled_expression_cache,
failed_expression_cache) with a thread-safe LRU or time-aware cache (e.g., moka,
lru_time_cache, or an LRU wrapper around DashMap) and add eviction policies in
the ConsumerFilterManager constructor/new path, ensure cache lookups/inserts in
methods that currently access compiled_expression_cache and
failed_expression_cache follow the new API, and add a background cleanup task or
TTL metadata if using time-based eviction; keep stop() behavior but remove
reliance on it for memory reclamation.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 8046d518-dd84-43f6-bbae-ed2572befa50
📒 Files selected for processing (9)
rocketmq-broker/src/filter/commit_log_dispatcher_calc_bit_map.rsrocketmq-broker/src/filter/consumer_filter_data.rsrocketmq-broker/src/filter/expression_message_filter.rsrocketmq-broker/src/filter/manager/consumer_filter_manager.rsrocketmq-broker/src/filter/message_evaluation_context.rsrocketmq-broker/src/processor/client_manage_processor.rsrocketmq-broker/src/processor/pop_message_processor.rsrocketmq-broker/src/processor/pull_message_processor.rsrocketmq-filter/src/filter/sql_runtime.rs
| .cloned() | ||
| .map(Value::String) | ||
| .unwrap_or(Value::Null)), | ||
| ExprNode::Now => Ok(Value::Long(current_millis() as i64)), |
Contributor
There was a problem hiding this comment.
Make NOW stable within a single evaluation.
Each ExprNode::Now reads the clock independently, so NOW = NOW and NOW BETWEEN NOW AND NOW can change value inside one evaluate() call. Capture the timestamp once in SqlExpression::evaluate() or the evaluation context and reuse it for every NOW node in that invocation.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-filter/src/filter/sql_runtime.rs` at line 109, ExprNode::Now
currently calls current_millis() per node, so make NOW stable by capturing a
single timestamp at the start of an evaluation and reusing it for every
ExprNode::Now in that invocation: modify SqlExpression::evaluate() (or the
evaluation context passed into node evaluation) to compute let now_ts =
current_millis() once and then change ExprNode::Now handling to return
Value::Long(now_ts as i64) (or read now_ts from the shared context) so all NOW
nodes in one evaluate() use the same value.
Comment on lines
+137
to
+152
| ExprNode::InList { expr, items, negated } => { | ||
| let value = expr.evaluate(context)?; | ||
| if matches!(value, Value::Null) { | ||
| return Ok(Value::Null); | ||
| } | ||
|
|
||
| let mut matched = false; | ||
| for item in items { | ||
| if eval_compare_values(CompareOp::Eq, &value, item)? == Some(true) { | ||
| matched = true; | ||
| break; | ||
| } | ||
| } | ||
|
|
||
| Ok(Value::Boolean(if *negated { !matched } else { matched })) | ||
| } |
Contributor
There was a problem hiding this comment.
Preserve NULL semantics for IN / NOT IN.
parse_constant_value() admits NULL, but this loop drops None from eval_compare_values() and collapses x IN (1, NULL) / x NOT IN (1, NULL) to definite booleans when x misses every non-null item. Either reject NULL members during parsing or carry an unknown flag here and return Value::Null unless a definite match was found.
Also applies to: 736-744
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-filter/src/filter/sql_runtime.rs` around lines 137 - 152, The IN/NOT
IN branch (ExprNode::InList) currently ignores unknown (NULL) comparison results
from eval_compare_values and therefore collapses expressions like x IN (1, NULL)
incorrectly; modify the loop to track an "unknown" flag when eval_compare_values
returns None, set matched=true if Some(true), and after the loop return
Value::Null if no definite match and unknown is true; otherwise return the
boolean (applying negation when *negated). Reference ExprNode::InList and
eval_compare_values (and alternatively parse_constant_value if you prefer
rejecting NULL members at parse time).
Comment on lines
+192
to
+206
| if search.is_empty() { | ||
| return Ok(Value::Boolean(false)); | ||
| } | ||
|
|
||
| let Value::String(subject) = expr.evaluate(context)? else { | ||
| return Ok(Value::Boolean(false)); | ||
| }; | ||
|
|
||
| let matched = match op { | ||
| StringMatchOp::Contains => subject.contains(search.as_str()), | ||
| StringMatchOp::StartsWith => subject.starts_with(search.as_str()), | ||
| StringMatchOp::EndsWith => subject.ends_with(search.as_str()), | ||
| }; | ||
|
|
||
| Ok(Value::Boolean(if *negated { !matched } else { matched })) |
Contributor
There was a problem hiding this comment.
Apply negated after the empty-pattern fast path.
The early return on search.is_empty() bypasses negation, so a NOT CONTAINS '' / NOT STARTSWITH '' / NOT ENDSWITH '' return false while NOT (a CONTAINS '') returns true. Compute the base result first and only then invert it, or reject empty RHS literals during parsing.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-filter/src/filter/sql_runtime.rs` around lines 192 - 206, The early
return on search.is_empty() bypasses negation; change the logic so you compute
the base match first and then apply negation: instead of returning
Ok(Value::Boolean(false)) when search.is_empty(), set matched = true (or
otherwise compute the same base result you would for an empty pattern) and then
return Ok(Value::Boolean(if *negated { !matched } else { matched })); adjust the
block around expr.evaluate(context) /
StringMatchOp::{Contains,StartsWith,EndsWith} to handle empty search as a
regular match case (or alternatively enforce non-empty RHS at parse time).
Comment on lines
+279
to
+299
| fn eval_compare_values(op: CompareOp, left: &Value, right: &Value) -> Result<Option<bool>, EvaluationError> { | ||
| if matches!(left, Value::Null) || matches!(right, Value::Null) { | ||
| return Ok(None); | ||
| } | ||
|
|
||
| let left = stringify_value(&left); | ||
| let right = stringify_value(&right); | ||
| let order = left.cmp(&right); | ||
| if let (Some(left), Some(right)) = (coerce_numeric(left), coerce_numeric(right)) { | ||
| let ordering = left.partial_cmp(&right).ok_or_else(|| { | ||
| EvaluationError::InvalidOperation(CheetahString::from_static_str( | ||
| "numeric comparison returned no ordering", | ||
| )) | ||
| })?; | ||
|
|
||
| return Ok(Some(match op { | ||
| CompareOp::Eq => numeric_eq(left, right), | ||
| CompareOp::Ne => !numeric_eq(left, right), | ||
| CompareOp::Gt => ordering == Ordering::Greater, | ||
| CompareOp::Gte => matches!(ordering, Ordering::Greater | Ordering::Equal), | ||
| CompareOp::Lt => ordering == Ordering::Less, | ||
| CompareOp::Lte => matches!(ordering, Ordering::Less | Ordering::Equal), | ||
| })); | ||
| } |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
python - <<'PY'
a = 9223372036854775807
b = 9223372036854775806
print(a == b)
print(float(a) == float(b))
print(float(a), float(b))
PYRepository: mxsm/rocketmq-rust
Length of output: 116
🏁 Script executed:
find . -name "sql_runtime.rs" -path "*/rocketmq-filter/*" | head -5Repository: mxsm/rocketmq-rust
Length of output: 105
🏁 Script executed:
wc -l rocketmq-filter/src/filter/sql_runtime.rsRepository: mxsm/rocketmq-rust
Length of output: 108
🏁 Script executed:
sed -n '279,299p' rocketmq-filter/src/filter/sql_runtime.rsRepository: mxsm/rocketmq-rust
Length of output: 1039
🏁 Script executed:
sed -n '347,365p' rocketmq-filter/src/filter/sql_runtime.rsRepository: mxsm/rocketmq-rust
Length of output: 655
Keep integral comparisons in i64 domain to prevent silent precision loss.
coerce_numeric() unconditionally converts Value::Long and integer strings to f64, causing integers above 2^53 to collapse to identical floating-point representations. This breaks equality and range comparisons for large integral values. Compare i64 operands directly before promoting to f64 only when a real floating-point operand is involved.
Confirmed with the following:
- Two distinct i64 values (9223372036854775807 and 9223372036854775806) both convert to 9.223372036854776e+18 as f64, making them indistinguishable for comparison.
- All numeric comparison paths in
eval_compare_values(lines 285–297) operate entirely on f64 results fromcoerce_numeric.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-filter/src/filter/sql_runtime.rs` around lines 279 - 299, The
numeric comparison currently uses coerce_numeric to convert everything to f64
(in eval_compare_values), causing precision loss for large i64 values; change
the logic in eval_compare_values so that if both operands are integral
(Value::Long or integer-string) you compare them as i64 directly (using i64
partial_cmp / equality) and only fall back to f64 comparisons when at least one
operand is a floating-point value; update or add helper coercion behavior
(coerce_numeric or a new coerce_to_numeric_enum) so it returns a discriminated
numeric kind (e.g., Int(i64) vs Float(f64)) and use that to call numeric_eq or
ordering appropriately, keeping existing EvaluationError paths (and using
numeric_eq for float branch) and preserving the existing early-null handling.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #7051 +/- ##
==========================================
+ Coverage 56.59% 56.72% +0.13%
==========================================
Files 1078 1078
Lines 178928 179430 +502
==========================================
+ Hits 101262 101783 +521
+ Misses 77666 77647 -19 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
rocketmq-rust-bot
approved these changes
Apr 4, 2026
rocketmq-rust-bot
left a comment
rocketmq-rust-bot
left a comment
Collaborator
There was a problem hiding this comment.
LGTM - All CI checks passed ✅
rocketmq-rust-bot
added
approved
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which Issue(s) This PR Fixes(Closes)
Brief Description
How Did You Test This Change?
Summary by CodeRabbit
Release Notes
New Features
IN/NOT IN,BETWEEN/NOT BETWEEN,CONTAINS,STARTSWITH,ENDSWITH, andNOW()function.Improvements