Fix intersection creating many duplicate events (#359)

koralium · Feb 11, 2024 · f446fad · f446fad
1 parent c001f0f
commit f446fad
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 1 deletion.
diff --git a/src/FlowtideDotNet.Base/Vertices/FixedPoint/FixedPointVertex.cs b/src/FlowtideDotNet.Base/Vertices/FixedPoint/FixedPointVertex.cs
@@ -164,7 +164,7 @@ private async IAsyncEnumerable<KeyValuePair<int, IStreamEvent>> OnLockingPrepare
 
         public void CreateBlock()
         {
-            singleReadSource = NoReadSourceInLoop();
+            singleReadSource = false;
 
             _transformBlock = new TransformManyBlock<KeyValuePair<int, IStreamEvent>, KeyValuePair<int, IStreamEvent>>((r) =>
             {

diff --git a/src/FlowtideDotNet.Connector.OpenFGA/Internal/ZanzibarToFlowtideVisitor.cs b/src/FlowtideDotNet.Connector.OpenFGA/Internal/ZanzibarToFlowtideVisitor.cs
@@ -163,6 +163,52 @@ public override Relation VisitZanzibarJoinIntersectWildcard(ZanzibarJoinIntersec
                                 },
                             }
                         },
+                        // l.object_type = r.object_type
+                        new ScalarFunction()
+                        {
+                            ExtensionUri = FunctionsComparison.Uri,
+                            ExtensionName = FunctionsComparison.Equal,
+                            Arguments = new List<Expression>()
+                            {
+                                new DirectFieldReference()
+                                {
+                                    ReferenceSegment = new StructReferenceSegment()
+                                    {
+                                        Field = ObjectTypeColumn
+                                    }
+                                },
+                                new DirectFieldReference()
+                                {
+                                    ReferenceSegment = new StructReferenceSegment()
+                                    {
+                                        Field = left.OutputLength + ObjectTypeColumn
+                                    }
+                                },
+                            }
+                        },
+                        // l.object_id = r.object_id
+                        new ScalarFunction()
+                        {
+                            ExtensionUri = FunctionsComparison.Uri,
+                            ExtensionName = FunctionsComparison.Equal,
+                            Arguments = new List<Expression>()
+                            {
+                                new DirectFieldReference()
+                                {
+                                    ReferenceSegment = new StructReferenceSegment()
+                                    {
+                                        Field = ObjectIdColumn
+                                    }
+                                },
+                                new DirectFieldReference()
+                                {
+                                    ReferenceSegment = new StructReferenceSegment()
+                                    {
+                                        Field = left.OutputLength + ObjectIdColumn
+                                    }
+                                },
+                            }
+                        },
                         // l.user_id = '*'
                         new ScalarFunction()
                         {