Repository: numaproj/numaflow
Pull request #394 (2022-11-30 09:48:35)

chore:non keyed stream support for reduce
Signed-off-by: Yashash H L <yashashhl25@gmail.com>

yhl25: I have tested this using a simple non-keyed reduce pipeline 
![Screenshot 2022-11-30 at 3 14 55 PM](https://user-images.githubusercontent.com/109710325/204768169-9fdbfb72-c930-405c-a35c-b8eae64da4e2.png)


Patch:
diff --git a/pkg/apis/numaflow/v1alpha1/const.go b/pkg/apis/numaflow/v1alpha1/const.go
index e71c8710..06fa0836 100644
--- a/pkg/apis/numaflow/v1alpha1/const.go
+++ b/pkg/apis/numaflow/v1alpha1/const.go
@@ -137,6 +137,9 @@ const (
 	DefaultStoreSyncDuration  = 2 * time.Second        // Default sync duration for pbq
 	DefaultStoreMaxBufferSize = 100000                 // Default buffer size for pbq in bytes
 	DefaultStorePath          = PathPBQMount + "/wals" // Default store path
+
+	// DefaultKeyForNonKeyedData Default key for non keyed stream
+	DefaultKeyForNonKeyedData = "NON_KEYED_STREAM"
 )
 
 var (
diff --git a/pkg/reduce/reduce.go b/pkg/reduce/reduce.go
index dc9177b1..99dc21be 100644
--- a/pkg/reduce/reduce.go
+++ b/pkg/reduce/reduce.go
@@ -23,6 +23,7 @@ import (
 	"context"
 	"time"
 
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"go.uber.org/zap"
 
 	"github.com/numaproj/numaflow/pkg/window"
@@ -47,6 +48,7 @@ type DataForward struct {
 	windowingStrategy   window.Windower
 	opts                *Options
 	log                 *zap.SugaredLogger
+	keyed               bool
 }
 
 func NewDataForward(ctx context.Context,
@@ -57,7 +59,9 @@ func NewDataForward(ctx context.Context,
 	whereToDecider forward.ToWhichStepDecider,
 	fw fetch.Fetcher,
 	watermarkPublishers map[string]publish.Publisher,
-	windowingStrategy window.Windower, opts ...Option) (*DataForward, error) {
+	windowingStrategy window.Windower,
+	keyed bool,
+	opts ...Option) (*DataForward, error) {
 
 	options := DefaultOptions()
 
@@ -76,6 +80,7 @@ func NewDataForward(ctx context.Context,
 		watermarkPublishers: watermarkPublishers,
 		windowingStrategy:   windowingStrategy,
 		log:                 logging.FromContext(ctx),
+		keyed:               keyed,
 		opts:                options}, nil
 }
 
@@ -149,6 +154,10 @@ func (d *DataForward) forwardAChunk(ctx context.Context) {
 	// elements in the batch based on the watermark we fetch from 0th offset.
 	processorWM := d.watermarkFetcher.GetWatermark(readMessages[0].ReadOffset)
 	for _, m := range readMessages {
+		if !d.keyed {
+			m.Key = dfv1.DefaultKeyForNonKeyedData
+			m.Message.Key = dfv1.DefaultKeyForNonKeyedData
+		}
 		m.Watermark = time.Time(processorWM)
 	}
 
diff --git a/pkg/sinks/logger/log.go b/pkg/sinks/logger/log.go
index afbb341c..0383d25e 100644
--- a/pkg/sinks/logger/log.go
+++ b/pkg/sinks/logger/log.go
@@ -99,7 +99,7 @@ func (s *ToLog) Write(_ context.Context, messages []isb.Message) ([]isb.Offset,
 	prefix := "(" + s.GetName() + ")"
 	for _, message := range messages {
 		logSinkWriteCount.With(map[string]string{metricspkg.LabelVertex: s.name, metricspkg.LabelPipeline: s.pipelineName}).Inc()
-		log.Println(prefix, " Payload - ", string(message.Payload), " Start - ", message.StartTime.UnixMilli(), " End - ", message.EndTime.UnixMilli())
+		log.Println(prefix, " Payload - ", string(message.Payload), " Key - ", message.Key, " Start - ", message.StartTime.UnixMilli(), " End - ", message.EndTime.UnixMilli())
 	}
 	return nil, make([]error, len(messages))
 }
diff --git a/pkg/udf/reduce_udf.go b/pkg/udf/reduce_udf.go
index 8fc2b671..75f85ce8 100644
--- a/pkg/udf/reduce_udf.go
+++ b/pkg/udf/reduce_udf.go
@@ -158,7 +158,8 @@ func (u *ReduceUDFProcessor) Start(ctx context.Context) error {
 			opts = append(opts, reduce.WithReadBatchSize(int64(*x.ReadBatchSize)))
 		}
 	}
-	dataforwarder, err := reduce.NewDataForward(ctx, udfHandler, reader, writers, pbqManager, conditionalForwarder, fetchWatermark, publishWatermark, windower, opts...)
+	dataforwarder, err := reduce.NewDataForward(ctx, udfHandler, reader, writers, pbqManager, conditionalForwarder,
+		fetchWatermark, publishWatermark, windower, u.VertexInstance.Vertex.Spec.UDF.GroupBy.Keyed, opts...)
 	if err != nil {
 		return fmt.Errorf("failed get a new DataForward, %w", err)
 	}


Test patch:
diff --git a/pkg/reduce/reduce_test.go b/pkg/reduce/reduce_test.go
index 7ffd5ddd..6dd035a6 100644
--- a/pkg/reduce/reduce_test.go
+++ b/pkg/reduce/reduce_test.go
@@ -26,6 +26,7 @@ import (
 	"testing"
 	"time"
 
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -41,6 +42,22 @@ import (
 	"github.com/numaproj/numaflow/pkg/window/strategy/fixed"
 )
 
+var keyedVertex = dfv1.Vertex{Spec: dfv1.VertexSpec{
+	PipelineName: "testPipeline",
+	AbstractVertex: dfv1.AbstractVertex{
+		Name: "testVertex",
+		UDF:  &dfv1.UDF{GroupBy: &dfv1.GroupBy{Keyed: true}},
+	},
+}}
+
+var nonKeyedVertex = dfv1.Vertex{Spec: dfv1.VertexSpec{
+	PipelineName: "testPipeline",
+	AbstractVertex: dfv1.AbstractVertex{
+		Name: "testVertex",
+		UDF:  &dfv1.UDF{GroupBy: &dfv1.GroupBy{Keyed: false}},
+	},
+}}
+
 type EventTypeWMProgressor struct {
 	watermarks map[string]processor.Watermark
 	m          sync.Mutex
@@ -215,7 +232,7 @@ func TestDataForward_StartWithNoOpWM(t *testing.T) {
 	window := fixed.NewFixed(windowTime)
 
 	var reduceDataForwarder *DataForward
-	reduceDataForwarder, err = NewDataForward(child, CounterReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, wmpublisher, publisher, window, WithReadBatchSize(10))
+	reduceDataForwarder, err = NewDataForward(child, CounterReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, wmpublisher, publisher, window, true, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	go reduceDataForwarder.Start(child)
@@ -284,7 +301,7 @@ func TestReduceDataForward_Count(t *testing.T) {
 
 	var reduceDataForward *DataForward
 	reduceDataForward, err = NewDataForward(ctx, CounterReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, p,
-		window, WithReadBatchSize(10))
+		window, true, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -357,7 +374,7 @@ func TestReduceDataForward_Sum(t *testing.T) {
 
 	var reduceDataForward *DataForward
 	reduceDataForward, err = NewDataForward(ctx, SumReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, p,
-		window, WithReadBatchSize(10))
+		window, true, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -430,7 +447,7 @@ func TestReduceDataForward_Max(t *testing.T) {
 
 	var reduceDataForward *DataForward
 	reduceDataForward, err = NewDataForward(ctx, MaxReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, p,
-		window, WithReadBatchSize(10))
+		window, true, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -503,7 +520,7 @@ func TestReduceDataForward_SumWithDifferentKeys(t *testing.T) {
 
 	var reduceDataForward *DataForward
 	reduceDataForward, err = NewDataForward(ctx, SumReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, p,
-		window, WithReadBatchSize(10))
+		window, true, WithReadBatchSize(10))
 	assert.NoError(t, err)
 
 	// start the forwarder
@@ -535,7 +552,7 @@ func TestReduceDataForward_SumWithDifferentKeys(t *testing.T) {
 	_ = json.Unmarshal(msgs[0].Payload, &readMessagePayload1)
 	_ = json.Unmarshal(msgs[1].Payload, &readMessagePayload2)
 	// since the window duration is 5 minutes, the output should be
-	// 100 * 300(for key even) and 100 * 97(for key odd)
+	// 100 * 300(for key even) and 99 * 300(for key odd)
 	// we cant guarantee the order of the output
 	assert.Contains(t, []int{30000, 29700}, readMessagePayload1.Value)
 	assert.Contains(t, []int{30000, 29700}, readMessagePayload2.Value)
@@ -544,6 +561,81 @@ func TestReduceDataForward_SumWithDifferentKeys(t *testing.T) {
 
 }
 
+// Max operation with 5 minutes window and non keyed
+func TestReduceDataForward_NonKeyed(t *testing.T) {
+	var (
+		ctx, cancel    = context.WithTimeout(context.Background(), 5*time.Second)
+		fromBufferSize = int64(100000)
+		toBufferSize   = int64(10)
+		messages       = []int{100, 99}
+		startTime      = 0 // time in millis
+		fromBufferName = "source-reduce-buffer"
+		toBufferName   = "reduce-to-buffer"
+		err            error
+	)
+
+	defer cancel()
+
+	// create from buffers
+	fromBuffer := simplebuffer.NewInMemoryBuffer(fromBufferName, fromBufferSize)
+
+	// create to buffers
+	buffer := simplebuffer.NewInMemoryBuffer(toBufferName, toBufferSize)
+	toBuffer := map[string]isb.BufferWriter{
+		toBufferName: buffer,
+	}
+
+	// create pbq manager
+	var pbqManager *pbq.Manager
+	pbqManager, err = pbq.NewManager(ctx, memory.NewMemoryStores(memory.WithStoreSize(1000)),
+		pbq.WithReadTimeout(1*time.Second), pbq.WithChannelBufferSize(10))
+	assert.NoError(t, err)
+
+	// create in memory watermark publisher and fetcher
+	f, p := fetcherAndPublisher(ctx, toBuffer, fromBuffer, t.Name())
+
+	// create a fixed window of 5 minutes
+	window := fixed.NewFixed(5 * time.Minute)
+
+	var reduceDataForward *DataForward
+	reduceDataForward, err = NewDataForward(ctx, SumReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, p,
+		window, false, WithReadBatchSize(10))
+	assert.NoError(t, err)
+
+	// start the forwarder
+	go reduceDataForward.Start(ctx)
+
+	// start the producer
+	go publishMessages(ctx, startTime, messages, 600, 10, p[fromBuffer.GetName()], fromBuffer)
+
+	// wait until there is data in to buffer
+	for buffer.IsEmpty() {
+		select {
+		case <-ctx.Done():
+			assert.Fail(t, ctx.Err().Error())
+			return
+		default:
+			time.Sleep(100 * time.Millisecond)
+		}
+	}
+
+	// we are reading only one message here but the count should be equal to
+	// the number of keyed windows that closed
+	msgs, readErr := buffer.Read(ctx, 1)
+	assert.Nil(t, readErr)
+	assert.Len(t, msgs, 1)
+
+	// assert the output of reduce
+	var readMessagePayload PayloadForTest
+	_ = json.Unmarshal(msgs[0].Payload, &readMessagePayload)
+	// since the window duration is 5 minutes, the output should be
+	// 100 * 300 + 99 * 300
+	// we cant guarantee the order of the output
+	assert.Equal(t, 59700, readMessagePayload.Value)
+	assert.Equal(t, "sum", readMessagePayload.Key)
+
+}
+
 func TestDataForward_WithContextClose(t *testing.T) {
 	var (
 		ctx, cancel    = context.WithTimeout(context.Background(), 5*time.Second)
@@ -587,7 +679,7 @@ func TestDataForward_WithContextClose(t *testing.T) {
 
 	var reduceDataForward *DataForward
 	reduceDataForward, err = NewDataForward(cctx, SumReduceTest{}, fromBuffer, toBuffer, pbqManager, CounterReduceTest{}, f, p,
-		window, WithReadBatchSize(1))
+		window, true, WithReadBatchSize(1))
 	assert.NoError(t, err)
 
 	// start the forwarder