feat: add Accumulator #237

cosmic-chichu · 2025-07-21T00:09:24Z

Fixes: #219

Implement accumulator
add tests

Signed-off-by: Sidhant Kohli <[email protected]>

- Fixed dictionary iteration issue in stream_send_eof method that was causing 'RuntimeError: dictionary changed size during iteration' - Fixed None watermark comparison issues in write_to_global_queue method - Fixed None watermark/event_time conversion to protobuf timestamp - Added proper null checks for watermark and event_time handling - Tests now run without runtime errors and produce expected message count The main fixes: 1. Create copy of dictionary keys before iteration in stream_send_eof 2. Add null checks before watermark comparison and protobuf conversion 3. Handle None values properly in Message watermark/event_time fields Tests are now passing the core functionality (message processing) but EOF handling still needs work. Signed-off-by: srao12 <[email protected]>

Signed-off-by: Sidhant Kohli <[email protected]>

- Fix missing default value in AccumulatorRequest constructor - Add watermark parameter handling in AccumulatorRequest - Improve test structure and remove debug test file - Fix task_manager.py to handle EOF counting properly - All 253 tests now pass successfully Signed-off-by: srao12 <[email protected]>

Signed-off-by: srao12 <[email protected]>

codecov · 2025-07-21T00:10:23Z

Codecov Report

❌ Patch coverage is 90.45346% with 40 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (main@7faf39b). Learn more about missing BASE report.
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
pynumaflow/accumulator/servicer/task_manager.py	80.14%	20 Missing and 8 partials ⚠️
pynumaflow/accumulator/servicer/async_servicer.py	79.48%	7 Missing and 1 partial ⚠️
pynumaflow/accumulator/async_server.py	90.90%	1 Missing and 2 partials ⚠️
pynumaflow/accumulator/_dtypes.py	99.50%	1 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff           @@
##             main     #237   +/-   ##
=======================================
  Coverage        ?   93.70%           
=======================================
  Files           ?       65           
  Lines           ?     2860           
  Branches        ?      150           
=======================================
  Hits            ?     2680           
  Misses          ?      130           
  Partials        ?       50

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

Signed-off-by: srao12 <[email protected]>

vigith

there a lot of typos, please do a self review?

examples/accumulator/streamsorter/Dockerfile

examples/accumulator/streamsorter/example.py

pynumaflow/accumulator/async_server.py

Signed-off-by: srao12 <[email protected]>

vigith

the proto file is stale

pynumaflow/accumulator/_dtypes.py

pynumaflow/proto/accumulator/accumulator.proto

Signed-off-by: srao12 <[email protected]>

examples/accumulator/streamsorter/Makefile

Signed-off-by: srao12 <[email protected]>

pynumaflow/accumulator/async_server.py

kohlisid · 2025-07-21T20:44:14Z

pynumaflow/accumulator/servicer/task_manager.py

+
+        if curr_task:
+            await self.tasks[unified_key].iterator.put(STREAM_EOF)
+            self.tasks.pop(unified_key)


Can you test a couple of small things for me please,

Before calling the pop on the the task, can you await on the _consumer_future task for this. Basically, check if it has completed the write_to_global queue task which sends the eof back.
Actually first you can wait for the future of the task
Then await for the consumer task
And after that we can consider this task done, and pop it out
@cosmic-chichu @vigith

# Once this is done, we know that the task has written all the results # to the local result queue fut = task.future await fut # # Send an EOF message to the local result queue # # This will signal that the task has completed processing await task.result_queue.put(STREAM_EOF) # Wait for the local queue to write # all the results of this task to the global result queue con_future = task.consumer_future await con_future

Signed-off-by: srao12 <[email protected]>

pynumaflow/accumulator/servicer/task_manager.py

Signed-off-by: srao12 <[email protected]>

- Add tests for OPEN, APPEND, and CLOSE window operations - Test standard flow (OPEN→APPEND→CLOSE), append-only, and mixed operation patterns - Add multi-key processing and error handling test scenarios Signed-off-by: srao12 <[email protected]>

Signed-off-by: srao12 <[email protected]>

cosmic-chichu · 2025-07-25T08:30:46Z

FMEA Test Report

Test Duration: 1200 seconds

Total Checks: 80

Pods Killed: 19

Python Sink Ordering Failures: 0

Go Sink Ordering Failures: 0

Result: SUCCESS

20 min fmea test: 10 tps, 1 key, panic every 80 seconds.
Event time ordering maintained throughout test!

High Throughput FMEA Test Report

Test Configuration

Duration: 5 minutes
Target: 50 keys, ~300 TPS
Pod Kills: 7

Results Summary

5 min high throughput fmea test: 300tps, 50 keys, no panics introduced, frequent pod kills.

Python Sink

Keys Analyzed: 50
Events Processed: 44945
Ordering Failures: 0

pynumaflow/accumulator/_dtypes.py

pynumaflow/accumulator/servicer/async_servicer.py

pynumaflow/accumulator/servicer/task_manager.py

tests/accumulator/test_async_accumulator_err.py

tests/accumulator/test_async_accumulator.py

kohlisid · 2025-07-25T19:17:19Z

pynumaflow/accumulator/_dtypes.py

+
+
+@dataclass(init=False)
+class Metadata:


The accumator signature doesn't expect metadata correct? Please confirm and remove if not required
https://github.com/numaproj/numaflow-go/blob/main/pkg/accumulator/examples/streamsorter/main.go#L23

removed it, not required

pynumaflow/accumulator/_dtypes.py

tests/accumulator/test_datatypes.py

kohlisid · 2025-07-25T19:23:37Z

@cosmic-chichu Did first pass, please check once

Signed-off-by: srao12 <[email protected]>

kohlisid · 2025-07-30T07:16:26Z

tests/accumulator/test_datatypes.py

+        self.assertEqual(2, r.test2)
+        self.assertEqual(1, r.test3)
+
+    def test_deep_copy(self):


Do we need to test deep copy here?

kohlisid · 2025-07-30T07:23:27Z

tests/accumulator/test_use_cases.py

+                    yield datum
+
+            # Process the stream through our accumulator
+            await sorter.handler(data_stream(), output)


This function is draining the iterator and sending a sorted order. Not sure if this covers accumulator functionality here much as we are calling the function directly without any RPC interations.

I guess this file can be removed in favor of actual end to end tests if that is the intention here

kohlisid · 2025-07-30T07:24:55Z

tests/accumulator/test_use_cases.py

+        self.assertIsNotNone(server.servicer)
+
+        # Example 2: Accumulator with configuration parameters
+        class ConfigurableAccumulator(Accumulator):


Same as above
I guess we can remove this as well

kohlisid

@cosmic-chichu
Please check the comments above and remove those redundant tests as well before merging

As next steps after the PR lets enable the e2e tests for accumulator on the mainline numaflow repo

We would need to add the accum example path here for stream sorter if thats used in the e2e
https://github.com/numaproj/numaflow-python/blob/main/.github/workflows/build-push.yaml#L20

Signed-off-by: srao12 <[email protected]>

kohlisid and others added 15 commits March 24, 2025 14:02

add proto

266d9d3

Signed-off-by: Sidhant Kohli <[email protected]>

init accumulator

96b8c73

Signed-off-by: Sidhant Kohli <[email protected]>

Merge branch 'main' into accum

67f5e87

minor fixes

8b58b63

Signed-off-by: Sidhant Kohli <[email protected]>

3/5 accum test files pass

c96b488

chore: add example

fa8fa6d

Signed-off-by: Sidhant Kohli <[email protected]>

merge

b49d53c

Signed-off-by: srao12 <[email protected]>

Fix merge conflicts

19d61a3

Signed-off-by: srao12 <[email protected]>

fix: example

10ebf46

Signed-off-by: srao12 <[email protected]>

fix: tests and add e2e test

32fda32

Signed-off-by: srao12 <[email protected]>

fix: cleanup logs

b4d7ccc

Signed-off-by: srao12 <[email protected]>

fix: resolve conflicts

6466f31

Signed-off-by: srao12 <[email protected]>

fix: use optimized Dockerfile

35156ba

Signed-off-by: srao12 <[email protected]>

cosmic-chichu self-assigned this Jul 21, 2025

cosmic-chichu requested review from ab93, vigith and whynowy as code owners July 21, 2025 00:09

fix: lint

ea5576c

Signed-off-by: srao12 <[email protected]>

vigith requested changes Jul 21, 2025

View reviewed changes

srao12 added 4 commits July 20, 2025 18:24

fix: tests and lint

54c0224

Signed-off-by: srao12 <[email protected]>

fix: update example

1d2863c

Signed-off-by: srao12 <[email protected]>

fix: tests

82f83d5

Signed-off-by: srao12 <[email protected]>

fix: lint

ee3acf9

Signed-off-by: srao12 <[email protected]>

vigith requested changes Jul 21, 2025

View reviewed changes

pynumaflow/accumulator/_dtypes.py Show resolved Hide resolved

pynumaflow/proto/accumulator/accumulator.proto Show resolved Hide resolved

srao12 added 2 commits July 20, 2025 20:46

fix: update proto

6b76b9d

Signed-off-by: srao12 <[email protected]>

fix: update docstring

5774e80

Signed-off-by: srao12 <[email protected]>

vigith mentioned this pull request Jul 21, 2025

[WIP] feat: add Accumulator #223

Closed

kohlisid reviewed Jul 21, 2025

View reviewed changes

examples/accumulator/streamsorter/Makefile Outdated Show resolved Hide resolved

fix: update docker image name

7cb73d7

Signed-off-by: srao12 <[email protected]>

kohlisid reviewed Jul 21, 2025

View reviewed changes

pynumaflow/accumulator/async_server.py Show resolved Hide resolved

kohlisid reviewed Jul 21, 2025

View reviewed changes

fix: update close task

d6968b6

Signed-off-by: srao12 <[email protected]>

kohlisid reviewed Jul 22, 2025

View reviewed changes

pynumaflow/accumulator/servicer/task_manager.py Outdated Show resolved Hide resolved

srao12 added 3 commits July 24, 2025 19:00

fix: tests

8aebf04

Signed-off-by: srao12 <[email protected]>

fix: conflicts

0156e24

Signed-off-by: srao12 <[email protected]>

kohlisid reviewed Jul 25, 2025

View reviewed changes

pynumaflow/accumulator/_dtypes.py Show resolved Hide resolved

kohlisid requested changes Jul 25, 2025

View reviewed changes

srao12 added 3 commits July 26, 2025 19:52

fix: comments and tests

6d8399a

Signed-off-by: srao12 <[email protected]>

fix: remove extra STREAM_EOF

53ec8da

Signed-off-by: srao12 <[email protected]>

fix: lint

e1a7c9a

Signed-off-by: srao12 <[email protected]>

kohlisid reviewed Jul 30, 2025

View reviewed changes

kohlisid approved these changes Jul 30, 2025

View reviewed changes

srao12 added 3 commits July 30, 2025 00:47

fix: remove tests

42c0264

Signed-off-by: srao12 <[email protected]>

fix: lint

be9c47f

Signed-off-by: srao12 <[email protected]>

fix: add accumulator for e2e tests

597638f

Signed-off-by: srao12 <[email protected]>

vigith approved these changes Jul 30, 2025

View reviewed changes

vigith merged commit 4982414 into numaproj:main Jul 30, 2025
11 checks passed

feat: add Accumulator #237

feat: add Accumulator #237

Uh oh!

Conversation

cosmic-chichu commented Jul 21, 2025 • edited by vigith Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov bot commented Jul 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

vigith left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

vigith left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kohlisid Jul 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kohlisid Jul 21, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

cosmic-chichu commented Jul 25, 2025

FMEA Test Report

Result: SUCCESS

High Throughput FMEA Test Report

Test Configuration

Results Summary

Python Sink

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

kohlisid Jul 25, 2025

Choose a reason for hiding this comment

Uh oh!

cosmic-chichu Jul 27, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

kohlisid commented Jul 25, 2025

Uh oh!

kohlisid Jul 30, 2025

Choose a reason for hiding this comment

Uh oh!

kohlisid Jul 30, 2025

Choose a reason for hiding this comment

Uh oh!

kohlisid Jul 30, 2025

Choose a reason for hiding this comment

Uh oh!

kohlisid left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

cosmic-chichu commented Jul 21, 2025 •

edited by vigith

Loading

codecov bot commented Jul 21, 2025 •

edited

Loading

kohlisid Jul 21, 2025 •

edited

Loading

kohlisid left a comment •

edited

Loading