Bug fix: Fix case where a post join condition could cause an early exit of the loop (#711)

* Bug fix: Fix case where a post join condition could cause an early exit of the loop

* change test to use modulo to future proof that it wont be able to use an index and only post join filter

Author: Ulimo

src/FlowtideDotNet.Core/Operators/Join/MergeJoin/ColumnStoreMergeJoin.cs

@@ -230,8 +230,7 @@ private async IAsyncEnumerable<StreamEventBatch> OnRecieveLeft(StreamEventBatch
                             {
                                 if (!_postCondition(columnReference.referenceBatch, columnReference.RowIndex, pageKeyStorage._data, k))
                                 {
-                                    _searchRightComparer.end = int.MinValue;
-                                    break;
+                                    continue;
                                 }
                             }
                             pageUpdated |= RecieveLeftHandleElement(
@@ -436,8 +435,7 @@ private async IAsyncEnumerable<StreamEventBatch> OnRecieveRight(StreamEventBatch
                             {
                                 if (!_postCondition(pageKeyStorage._data, k, columnReference.referenceBatch, columnReference.RowIndex))
                                 {
-                                    _searchLeftComparer.end = int.MinValue;
-                                    break;
+                                    continue;
                                 }
                             }
                             pageUpdated |= RecieveRightHandleElement(
@@ -503,6 +501,7 @@ private async IAsyncEnumerable<StreamEventBatch> OnRecieveRight(StreamEventBatch
                         {
                             await page.SavePage(false);
                         }
+                        // Checks if we end before the last element, if we do, we dont have to fetch the next page
                         if (_searchLeftComparer.end < (page.Keys.Count - 1))
                         {
                             break;

tests/FlowtideDotNet.AcceptanceTests/FlowtideAcceptanceBase.cs

@@ -120,6 +120,16 @@ public void AddOrUpdateOrder(Order order)
             flowtideTestStream.AddOrUpdateOrder(order);
         }
 
+        public void AddOrUpdateProject(Project project)
+        {
+            flowtideTestStream.AddOrUpdateProject(project);
+        }
+
+        public void AddOrUpdateProjectMember(ProjectMember projectMember)
+        {
+            flowtideTestStream.AddOrUpdateProjectMember(projectMember);
+        }
+
         public void DeleteUser(User user)
         {
             flowtideTestStream.DeleteUser(user);

tests/FlowtideDotNet.AcceptanceTests/Internal/DatasetGenerator.cs

@@ -85,6 +85,38 @@ public void AddOrUpdateOrder(Order order)
             mockTable.AddOrUpdate(new List<Order>() { order });
         }
 
+        public void AddOrUpdateProject(Project project)
+        {
+            var index = Projects.FindIndex(x => x.ProjectKey == project.ProjectKey);
+
+            if (index >= 0)
+            {
+                Projects[index] = project;
+            }
+            else
+            {
+                Projects.Add(project);
+            }
+            var mockTable = mockDatabase.GetOrCreateTable<Project>("projects");
+            mockTable.AddOrUpdate(new List<Project>() { project });
+        }
+
+        public void AddOrUpdateProjectMember(ProjectMember projectMember)
+        {
+            var index = ProjectMembers.FindIndex(x => x.ProjectMemberKey == projectMember.ProjectMemberKey);
+
+            if (index >= 0)
+            {
+                ProjectMembers[index] = projectMember;
+            }
+            else
+            {
+                ProjectMembers.Add(projectMember);
+            }
+            var mockTable = mockDatabase.GetOrCreateTable<ProjectMember>("projectmembers");
+            mockTable.AddOrUpdate(new List<ProjectMember>() { projectMember });
+        }
+
         public void AddOrUpdateCompany(Company company)
         {
             var index = Companies.FindIndex(x => x.CompanyId == company.CompanyId);

tests/FlowtideDotNet.AcceptanceTests/Internal/FlowtideTestStream.cs

@@ -155,6 +155,16 @@ public void AddOrUpdateOrder(Order order)
             generator.AddOrUpdateOrder(order);
         }
 
+        public void AddOrUpdateProject(Project project)
+        {
+            generator.AddOrUpdateProject(project);
+        }
+
+        public void AddOrUpdateProjectMember(ProjectMember projectMember)
+        {
+            generator.AddOrUpdateProjectMember(projectMember);
+        }
+
 
         [MemberNotNull(nameof(_connectorManager))]
         public void SetupConnectorManager()

tests/FlowtideDotNet.AcceptanceTests/JoinTests.cs

@@ -1402,5 +1402,45 @@ from subuser in gj.DefaultIfEmpty()
 
             AssertCurrentDataEqual(fullOuterJoin);
         }
+
+        [Fact]
+        public async Task PostJoinConditionShouldCheckAllRows()
+        {
+            // Add members first to make sure all the data is inside the tree
+            AddOrUpdateProjectMember(new ProjectMember()
+            {
+                CompanyId = "1",
+                ProjectNumber = "123",
+                UserKey = 0,
+                ProjectMemberKey = 1
+            });
+            AddOrUpdateProjectMember(new ProjectMember()
+            {
+                CompanyId = "1",
+                ProjectNumber = "123",
+                UserKey = 1,
+                ProjectMemberKey = 2
+            });
+            await StartStream(@"
+                INSERT INTO output 
+                SELECT 
+                    pm.userkey
+                FROM projects p
+                LEFT JOIN projectmembers pm
+                ON p.projectnumber = pm.projectnumber AND p.companyid = pm.companyid AND pm.userkey % ProjectKey = 1", ignoreSameDataCheck: false);
+            await WaitForUpdate();
+            // add the project to do the join
+            AddOrUpdateProject(new Project()
+            {
+                ProjectKey = 2,
+                CompanyId = "1",
+                ProjectNumber = "123",
+                Name = "Project 1"
+            });
+            await WaitForUpdate();
+
+            var actualData = GetActualRows();
+            AssertCurrentDataEqual(new[] { new { val = 1 } });
+        }
     }
 }