Implement count windows

[quentin-quix]

Implement sdf.tumbling_count_window. It creates a count-based tumbling window transformation on the StreamingDataFrame.

Tumbling windows divide messages into fixed-batch, non-overlapping windows. They allow performing stateful aggregations like sum, reduce, etc. on top of the data and emit results downstream.

This type of window close after a set count of messages.

app = Application()
sdf = app.dataframe(...)

sdf = (
    # Define a tumbling window of 10 messages
    sdf.tumbling_count_window(count=10)

    # Specify the aggregation function
    .sum()

    # Specify how the results should be emitted downstream.
    # "current()" will emit results as they come for each updated window,
    # possibly producing multiple messages per key-window pair
    # "final()" will emit windows only when they are closed and cannot
    # receive any updates anymore.
    .current()
)

[gwaramadze]

This will perform similarly to reduce because the whole list must be deserialized and re-serialized on every incoming message. You must use state.add_to_collection(value) instead.

Then to collect the value for windows closure you will use state._get_values method so it will have to be made "public" for this purpose.

[gwaramadze]

can you explain why is this constant 1 and 10?

[gwaramadze]

maybe this should be moved by the end of the method, together with state.delete_from_collection

[gwaramadze]

Also maybe lets name the method get_collection and variable also collection because value in this context has slightly different meaning.

Suggested change

	values = state.get_from_collection(start=1, end=10)
	collection = state.get_collection(start=1, end=10)

Given that, renaming _get_values to get_collection is a good move.