Fixing bug that dropped punctuations for Grouped Afa SingleEvent pipe when IsSyncTimeSimultaneityFree is true (#142)

Author: peterfreiling

Sources/Core/Microsoft.StreamProcessing/Operators/Afa/CompiledGroupedAfaPipe_SingleEvent.cs

@@ -320,10 +320,10 @@ public override unsafe void OnNext(StreamMessage<TKey, TPayload> batch)
                                 if (this.IsDeterministic) break; // We are guaranteed to have only one start state
                             }
                         }
-                        else if (batch.vother.col[i] < 0 && !this.IsSyncTimeSimultaneityFree)
+                        else if (batch.vother.col[i] < 0)
                         {
                             long synctime = src_vsync[i];
-                            if (synctime > this.lastSyncTime) // move time forward
+                            if (!this.IsSyncTimeSimultaneityFree && synctime > this.lastSyncTime) // move time forward
                             {
                                 this.seenEvent.Clear();
 

Sources/Test/SimpleTesting/Streamables/AfaTests.cs

@@ -844,16 +844,21 @@ public void GroupedAfa_IsSyncTimeSimultaneityFree()
         {
             var source = new StreamEvent<Tuple<string, int>>[]
             {
-                 StreamEvent.CreateStart(0, new Tuple<string, int>("A", 1)),
+                StreamEvent.CreateStart(0, new Tuple<string, int>("A", 1)),
                     StreamEvent.CreateStart(1, new Tuple<string, int>("A", 2)),
                     StreamEvent.CreateStart(1, new Tuple<string, int>("B", 2)),
-                 StreamEvent.CreateStart(3, new Tuple<string, int>("A", 1)),
-                 StreamEvent.CreateStart(4, new Tuple<string, int>("B", 1)),
+                StreamEvent.CreateStart(3, new Tuple<string, int>("A", 1)),
+
+                StreamEvent.CreatePunctuation<Tuple<string, int>>(4),
+
+                StreamEvent.CreateStart(4, new Tuple<string, int>("B", 1)),
                     StreamEvent.CreateStart(4, new Tuple<string, int>("B", 2)),
-                 StreamEvent.CreateStart(5, new Tuple<string, int>("B", 1)),
-                 StreamEvent.CreateStart(5, new Tuple<string, int>("C", 1)),
+                StreamEvent.CreateStart(5, new Tuple<string, int>("B", 1)),
+                StreamEvent.CreateStart(5, new Tuple<string, int>("C", 1)),
                     StreamEvent.CreateStart(6, new Tuple<string, int>("B", 2)),
                     StreamEvent.CreateStart(7, new Tuple<string, int>("A", 2)),
+
+                StreamEvent.CreatePunctuation<Tuple<string, int>>(7),
             }.ToObservable()
                 .ToStreamable()
                 .AlterEventDuration(10);
@@ -881,18 +886,21 @@ public void GroupedAfa_IsSyncTimeSimultaneityFree()
 
             var result = afa_compiled
                 .ToStreamEventObservable()
-                .Where(evt => evt.IsData)
                 .ToEnumerable()
                 .ToArray();
             var expected = new StreamEvent<Tuple<string, int>>[]
             {
                 StreamEvent.CreateInterval(1, 11, new Tuple<string, int>("AB", 2)),
+                StreamEvent.CreatePunctuation<Tuple<string, int>>(4),
                 StreamEvent.CreateInterval(4, 13, new Tuple<string, int>("AB", 1)),
                 StreamEvent.CreateInterval(4, 10, new Tuple<string, int>("AAB", 1)),
                 StreamEvent.CreateInterval(4, 11, new Tuple<string, int>("ABB", 2)),
                 StreamEvent.CreateInterval(5, 13, new Tuple<string, int>("ABB", 1)),
                 StreamEvent.CreateInterval(5, 10, new Tuple<string, int>("AABB", 1)),
                 StreamEvent.CreateInterval(6, 11, new Tuple<string, int>("ABBB", 2)),
+                StreamEvent.CreatePunctuation<Tuple<string, int>>(7),
+
+                StreamEvent.CreatePunctuation<Tuple<string, int>>(StreamEvent.InfinitySyncTime),
             };
             Assert.IsTrue(result.SequenceEqual(expected));
         }