Repository: numaproj/numaflow
Pull request #427 (2022-12-15 16:36:09)

test: sliding window e2e test


Patch:
diff --git a/examples/7-reduce-sliding-window.yaml b/examples/7-reduce-sliding-window.yaml
index acc18872..1d214bfa 100644
--- a/examples/7-reduce-sliding-window.yaml
+++ b/examples/7-reduce-sliding-window.yaml
@@ -6,7 +6,7 @@ metadata:
     loglevel: "true"
 spec:
   limits:
-    readBatchSize: 1
+    readBatchSize: 50
   vertices:
     - name: in
       containerTemplate:
@@ -30,9 +30,6 @@ spec:
     - name: reduce-sliding
       udf:
         container:
-          env:
-            - name: NUMAFLOW_DEBUG
-              value: "true"
           # compute the sum
           image: quay.io/numaio/go-integer-sum-example
         groupBy:
diff --git a/pkg/pbq/store/memory/stores.go b/pkg/pbq/store/memory/stores.go
index 8487376a..264e2bb9 100644
--- a/pkg/pbq/store/memory/stores.go
+++ b/pkg/pbq/store/memory/stores.go
@@ -36,7 +36,7 @@ type memoryStores struct {
 
 func NewMemoryStores(opts ...Option) store.StoreProvider {
 	s := &memoryStores{
-		storeSize:  100,
+		storeSize:  100000,
 		partitions: make(map[partition.ID]*memoryStore),
 	}
 
diff --git a/pkg/reduce/readloop/ordered.go b/pkg/reduce/readloop/ordered.go
index a7209105..16ce0450 100644
--- a/pkg/reduce/readloop/ordered.go
+++ b/pkg/reduce/readloop/ordered.go
@@ -66,6 +66,36 @@ func newOrderedForwarder(ctx context.Context) *orderedForwarder {
 	return of
 }
 
+func (of *orderedForwarder) insertTask(tx context.Context, t *task) {
+	of.Lock()
+	defer of.Unlock()
+
+	if of.taskQueue.Len() == 0 {
+		of.taskQueue.PushBack(t)
+		return
+	}
+
+	earliestTask := of.taskQueue.Front().Value.(*task)
+	recentTask := of.taskQueue.Back().Value.(*task)
+
+	// late arrival
+	if !earliestTask.pf.PartitionID.End.Before(t.pf.PartitionID.End) {
+		of.taskQueue.PushFront(t)
+	} else if !recentTask.pf.PartitionID.End.After(t.pf.PartitionID.End) {
+		// early arrival
+		of.taskQueue.PushBack(t)
+	} else {
+		// a task in the middle
+		for e := of.taskQueue.Back(); e != nil; e = e.Prev() {
+			_task := e.Value.(*task)
+			if !_task.pf.PartitionID.End.After(t.pf.PartitionID.End) {
+				of.taskQueue.InsertAfter(t, e)
+				break
+			}
+		}
+	}
+}
+
 // schedulePnF creates and schedules the PnF routine.
 func (of *orderedForwarder) schedulePnF(ctx context.Context,
 	udf applier.ReduceApplier,
@@ -82,9 +112,7 @@ func (of *orderedForwarder) schedulePnF(ctx context.Context,
 		pf:     pf,
 	}
 
-	of.Lock()
-	defer of.Unlock()
-	of.taskQueue.PushBack(t)
+	of.insertTask(ctx, t)
 
 	// invoke the reduce function
 	go of.reduceOp(ctx, t)
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index 6af0b63e..6e7def7d 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -23,9 +23,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
 	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
 	"github.com/numaproj/numaflow/pkg/window/strategy/sliding"
-
 	"go.uber.org/zap"
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
@@ -34,7 +34,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/pbq"
 	"github.com/numaproj/numaflow/pkg/pbq/store"
-	"github.com/numaproj/numaflow/pkg/pbq/store/memory"
 	"github.com/numaproj/numaflow/pkg/pbq/store/wal"
 	"github.com/numaproj/numaflow/pkg/reduce"
 	"github.com/numaproj/numaflow/pkg/shared/logging"


Test patch:
diff --git a/pkg/reduce/readloop/ordered_test.go b/pkg/reduce/readloop/ordered_test.go
index 40c8e417..0d4c984b 100644
--- a/pkg/reduce/readloop/ordered_test.go
+++ b/pkg/reduce/readloop/ordered_test.go
@@ -117,10 +117,9 @@ func TestOrderedProcessing(t *testing.T) {
 			}
 			assert.Equal(t, op.taskQueue.Len(), tt.expectedBefore)
 			count := 0
-			for e := op.taskQueue.Front(); e != nil; e = e.Next() {
+			for e := op.taskQueue.Back(); e != nil; e = e.Prev() {
 				pfTask := e.Value.(*task)
-				partitionKey := pfTask.pf.PartitionID
-				assert.Equal(t, partition.ID{Key: fmt.Sprintf("partition-%d", count)}, partitionKey)
+				assert.Equal(t, partition.ID{Key: fmt.Sprintf("partition-%d", count)}, pfTask.pf.PartitionID)
 				count = count + 1
 			}
 
diff --git a/test/reduce-e2e/reduce_test.go b/test/reduce-e2e/reduce_test.go
index 37810766..b860fd70 100644
--- a/test/reduce-e2e/reduce_test.go
+++ b/test/reduce-e2e/reduce_test.go
@@ -202,6 +202,40 @@ func (r *ReduceSuite) TestSimpleReducePipelineFailOverUsingWAL() {
 	w.Expect().VertexPodLogContains("sink", "Payload -  240", PodLogCheckOptionWithCount(1))
 }
 
+// two reduce vertex(keyed and non keyed) followed by a sliding window vertex
+func (r *ReduceSuite) TestComplexSlidingWindowPipeline() {
+	w := r.Given().Pipeline("@testdata/complex-sliding-window-pipeline.yaml").
+		When().
+		CreatePipelineAndWait()
+
+	defer w.DeletePipelineAndWait()
+
+	// wait for all the pods to come up
+	w.Expect().
+		VertexPodsRunning()
+
+	// port forward source vertex(to publish messages)
+	defer w.VertexPodPortForward("in", 8443, dfv1.VertexHTTPSPort).
+		TerminateAllPodPortForwards()
+
+	// publish messages to source vertex, with event time starting from 60000
+	startTime := 60000
+	for i := 0; i < 300; i++ {
+		eventTime := strconv.Itoa(startTime + i*1000)
+
+		HTTPExpect(r.T(), "https://localhost:8443").POST("/vertices/in").WithBytes([]byte("1")).WithHeader("X-Numaflow-Event-Time", eventTime).
+			Expect().
+			Status(204)
+		HTTPExpect(r.T(), "https://localhost:8443").POST("/vertices/in").WithBytes([]byte("2")).WithHeader("X-Numaflow-Event-Time", eventTime).
+			Expect().
+			Status(204)
+	}
+
+	w.Expect().VertexPodLogContains("sink", "Payload -  15  Key -  NON_KEYED_STREAM  Start -  20000  End -  80000")
+	w.Expect().VertexPodLogContains("sink", "Payload -  45  Key -  NON_KEYED_STREAM  Start -  30000  End -  90000")
+	w.Expect().VertexPodLogContains("sink", "Payload -  180  Key -  NON_KEYED_STREAM  Start -  80000  End -  140000")
+}
+
 func TestReduceSuite(t *testing.T) {
 	suite.Run(t, new(ReduceSuite))
 }
diff --git a/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml b/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
index 2c000c89..ef957825 100644
--- a/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/complex-reduce-pipeline.yaml
@@ -4,7 +4,7 @@ metadata:
   name: complex-sum
 spec:
   limits:
-    readBatchSize: 1
+    readBatchSize: 50
   vertices:
     - name: in
       scale:
@@ -18,7 +18,7 @@ spec:
         container:
           # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
           image: quay.io/numaio/go-even-odd-example
-    - name: first-aggregation
+    - name: keyed-fixed-sum
       udf:
         container:
           # compute the sum
@@ -28,7 +28,7 @@ spec:
             fixed:
               length: 10s
           keyed: true
-    - name: second-aggregation
+    - name: non-keyed-fixed-sum
       udf:
         container:
           # compute the sum
@@ -47,9 +47,9 @@ spec:
     - from: in
       to: atoi
     - from: atoi
-      to: first-aggregation
+      to: keyed-fixed-sum
       parallelism: 2
-    - from: first-aggregation
-      to: second-aggregation
-    - from: second-aggregation
+    - from: keyed-fixed-sum
+      to: non-keyed-fixed-sum
+    - from: non-keyed-fixed-sum
       to: sink
diff --git a/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml b/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
new file mode 100644
index 00000000..400b66d1
--- /dev/null
+++ b/test/reduce-e2e/testdata/complex-sliding-window-pipeline.yaml
@@ -0,0 +1,68 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: complex-sliding-sum
+spec:
+  limits:
+    readBatchSize: 50
+  vertices:
+    - name: in
+      scale:
+        min: 1
+      source:
+        http: {}
+    - name: atoi
+      scale:
+        min: 1
+      udf:
+        container:
+          # Tell the input number is even or odd, see https://github.com/numaproj/numaflow-go/tree/main/pkg/function/examples/evenodd
+          image: quay.io/numaio/go-even-odd-example
+    - name: keyed-fixed-sum
+      udf:
+        container:
+          # compute the sum
+          image: quay.io/numaio/go-integer-sum-example
+        groupBy:
+          window:
+            fixed:
+              length: 5s
+          keyed: true
+    - name: non-keyed-fixed-sum
+      udf:
+        container:
+          # compute the sum
+          image: quay.io/numaio/go-integer-sum-example
+        groupBy:
+          window:
+            fixed:
+              length: 10s
+          keyed: false
+    - name: non-keyed-sliding-sum
+      udf:
+        container:
+          # compute the sum
+          image: quay.io/numaio/go-integer-sum-example
+        groupBy:
+          window:
+            sliding:
+              length: 60s
+              slide: 10s
+          keyed: false
+    - name: sink
+      scale:
+        min: 1
+      sink:
+        log: {}
+  edges:
+    - from: in
+      to: atoi
+    - from: atoi
+      to: keyed-fixed-sum
+      parallelism: 2
+    - from: keyed-fixed-sum
+      to: non-keyed-fixed-sum
+    - from: non-keyed-fixed-sum
+      to: non-keyed-sliding-sum
+    - from: non-keyed-sliding-sum
+      to: sink
diff --git a/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml b/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
index 93b6b2c6..5f1a293c 100644
--- a/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/simple-keyed-reduce-pipeline.yaml
@@ -4,7 +4,7 @@ metadata:
   name: even-odd-sum
 spec:
   limits:
-    readBatchSize: 10
+    readBatchSize: 50
   vertices:
     - name: in
       scale:
diff --git a/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml b/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
index 4542f782..47924830 100644
--- a/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
+++ b/test/reduce-e2e/testdata/simple-non-keyed-reduce-pipeline.yaml
@@ -4,7 +4,7 @@ metadata:
   name: reduce-sum
 spec:
   limits:
-    readBatchSize: 10
+    readBatchSize: 50
   vertices:
     - name: in
       scale:
diff --git a/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml b/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
index d72662f2..cfdf3b20 100644
--- a/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
+++ b/test/reduce-e2e/testdata/simple-reduce-pipeline-wal.yaml
@@ -4,7 +4,7 @@ metadata:
   name: even-odd-sum
 spec:
   limits:
-    readBatchSize: 1
+    readBatchSize: 50
   vertices:
     - name: in
       source: