Adds Inner Join functionality to stream processing #93
Conversation
WalkthroughThe changes introduce an Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant StreamBuilder
participant JoinBuilder
participant JoinByKeyIntoBuilder
User->>StreamBuilder: Join(topic) / InnerJoin(topic)
StreamBuilder->>JoinBuilder: new JoinBuilder(..., innerJoin)
JoinBuilder->>JoinByKeyIntoBuilder: new JoinByKeyIntoBuilder(..., innerJoin)
JoinByKeyIntoBuilder->>JoinByKeyIntoBuilder: Handle(tuple)
Note right of JoinByKeyIntoBuilder: Behavior changes based on innerJoin flag
Poem
✨ Finishing Touches
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. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
src/MinimalKafka/Stream/Internals/JoinBuilder.cs (1)
10-11: Consider extending inner join support to predicate-based joins.The predicate-based
Onmethod createsJoinIntoBuilderwithout inner join support, while the key-basedOnmethod supports inner joins. This creates inconsistent functionality where inner joins only work with key-based joins.Consider whether
JoinIntoBuildershould also support inner join functionality for consistency.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
src/MinimalKafka/Stream/Internals/JoinBuilder.cs(2 hunks)src/MinimalKafka/Stream/Internals/JoinByKeyIntoBuilder.cs(2 hunks)src/MinimalKafka/Stream/Internals/StreamBuilder.cs(1 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (2)
src/MinimalKafka/Stream/Internals/JoinBuilder.cs (1)
src/MinimalKafka/Stream/Internals/JoinByKeyIntoBuilder.cs (1)
JoinByKeyIntoBuilder(6-75)
src/MinimalKafka/Stream/Internals/StreamBuilder.cs (2)
src/MinimalKafka/Stream/Internals/JoinBuilder.cs (1)
JoinBuilder(5-16)src/MinimalKafka/Stream/IStreamBuilder.cs (1)
IJoinBuilder(20-21)
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: Analyze
🔇 Additional comments (6)
src/MinimalKafka/Stream/Internals/JoinByKeyIntoBuilder.cs (2)
10-10: LGTM! Constructor parameter addition is correct.The
innerJoinparameter properly enables differentiation between join types in the constructor.
44-44: LGTM! Inner join logic correctly implemented.The conditional logic properly implements join semantics:
- Regular joins (innerJoin=false): Skip only when left value is null (left join behavior)
- Inner joins (innerJoin=true): Skip when either left OR right value is null (inner join behavior)
This correctly filters records based on the join type.
src/MinimalKafka/Stream/Internals/JoinBuilder.cs (2)
5-5: LGTM! Constructor parameter addition is correct.The
innerJoinparameter is properly added to enable join type differentiation.
15-15: LGTM! Parameter propagation is correct.The
innerJoinparameter is properly passed toJoinByKeyIntoBuilderto enable conditional join behavior.src/MinimalKafka/Stream/Internals/StreamBuilder.cs (2)
14-15: LGTM! Join method correctly maintains existing behavior.The explicit
falseparameter ensures backward compatibility by preserving the existing left join behavior for the regularJoinmethod.
17-21: LGTM! InnerJoin method provides clear API distinction.The new
InnerJoinmethod with explicittrueparameter provides a clean API for inner join functionality while maintaining consistency with the existingJoinmethod signature.
| public IJoinBuilder<TKey, TValue, K2, V2> InnerJoin<K2, V2>(string topic) | ||
| where K2 : notnull | ||
| { | ||
| return new JoinBuilder<TKey, TValue, K2, V2>(builder, _topic, topic, true); | ||
| } |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Verify interface definition includes InnerJoin method.
The new InnerJoin method is added to the concrete StreamBuilder class. Ensure that the IStreamBuilder<TKey, TValue> interface is updated to include this method, otherwise users won't be able to access inner join functionality when working with the interface.
🏁 Script executed:
#!/bin/bash
# Description: Check if IStreamBuilder interface includes InnerJoin method
# Expected: InnerJoin method should be declared in the interface
echo "Searching for IStreamBuilder interface definition..."
rg -A 10 "interface IStreamBuilder" --type cs
echo -e "\nSearching for InnerJoin method in interface files..."
rg "InnerJoin" --type cs -g "**/I*.cs"Length of output: 1610
Add InnerJoin to IStreamBuilder interface
The concrete StreamBuilder now exposes an InnerJoin<K2, V2>(string topic) method, but the IStreamBuilder<TKey, TValue> interface lacks this signature. Users programming against the interface won’t see or be able to call InnerJoin until it’s declared.
• File to update: src/MinimalKafka/Stream/IStreamBuilder.cs
• Insert alongside the existing join methods, for example after the current Join declaration:
public interface IStreamBuilder<K1, V1> : IIntoBuilder<K1, V1>
{
/// <summary>
/// Creates a join operation between the current stream and another stream specified by the given topic.
/// </summary>
IJoinBuilder<K1, V1, K2, V2> Join<K2, V2>(string topic)
where K2 : notnull;
+ /// <summary>
+ /// Creates an inner-join operation between the current stream and another stream specified by the given topic.
+ /// </summary>
+ IJoinBuilder<K1, V1, K2, V2> InnerJoin<K2, V2>(string topic)
+ where K2 : notnull;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| public IJoinBuilder<TKey, TValue, K2, V2> InnerJoin<K2, V2>(string topic) | |
| where K2 : notnull | |
| { | |
| return new JoinBuilder<TKey, TValue, K2, V2>(builder, _topic, topic, true); | |
| } | |
| // File: src/MinimalKafka/Stream/IStreamBuilder.cs | |
| public interface IStreamBuilder<K1, V1> : IIntoBuilder<K1, V1> | |
| { | |
| /// <summary> | |
| /// Creates a join operation between the current stream and another stream specified by the given topic. | |
| /// </summary> | |
| IJoinBuilder<K1, V1, K2, V2> Join<K2, V2>(string topic) | |
| where K2 : notnull; | |
| /// <summary> | |
| /// Creates an inner-join operation between the current stream and another stream specified by the given topic. | |
| /// </summary> | |
| IJoinBuilder<K1, V1, K2, V2> InnerJoin<K2, V2>(string topic) | |
| where K2 : notnull; | |
| } |
🤖 Prompt for AI Agents
In src/MinimalKafka/Stream/IStreamBuilder.cs around the lines where join methods
are declared, add the method signature for InnerJoin<K2, V2>(string topic) to
the IStreamBuilder<TKey, TValue> interface. This will ensure that users
programming against the interface can see and call the InnerJoin method exposed
by the StreamBuilder class. Place the new method declaration alongside existing
join methods, preferably after the current Join method declaration.
Adds the ability to perform inner joins between streams, allowing for records with matching keys to be combined.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
src/MinimalKafka/Stream/IStreamBuilder.cs (1)
16-20: Consider enhancing the documentation for better clarity.The new
InnerJoinmethod addition looks good and maintains consistency with the existingJoinmethod. However, the documentation could be more specific about inner join semantics - consider explaining how it differs from a regular join (e.g., "only returns results when both streams have matching keys")./// <summary> -/// Creates an inner-join operation between the current stream and another stream specified by the given topic. +/// Creates an inner-join operation between the current stream and another stream specified by the given topic. +/// Returns results only when both streams have matching keys, filtering out null values from the right stream. /// </summary>
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/MinimalKafka/Stream/IStreamBuilder.cs(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: Analyze
Adds the ability to perform inner joins in stream processing.
InnerJoinmethod toStreamBuilderto allow for creating inner joins.JoinByKeyIntoBuilderto only process records when both the left and right streams have a value, effectively implementing an inner join.Summary by CodeRabbit
New Features
Bug Fixes