Loadingone moment…
Sign in to run and submit your work
Reading is open to everyone. Running code and saving drafts need an account so your work is yours and comes back on your next visit.
or
CODE WORKSPACE
The streaming writer batches records before writing, under three limits: a maximum number of rows, a maximum size, and a maximum delay so a quiet stream still gets written. Tuning it is guesswork because nobody can see which limit is actually firing, and during quiet periods records sit in memory far longer than the delay suggests.
Write plan_flushes(records, max_rows, max_bytes, max_delay).
Function to write
plan_flushes(records: list[dict], max_rows: int, max_bytes: int, max_delay: float) -> list[dict]One entry per flush, with the reason, the row ids, the byte total and when the batch opened and closed.
How to approach it
Keep the batch's open time as state, and decide the flush reason before adding the record.
Sample cases
+ 2 held back until you submit
a burst then a quiet period
The row limit closes one batch, the delay closes another, and an oversized record forces its own.
Input
Argument 1
| id | at | bytes |
|---|---|---|
| m1 | 0 | 300 |
| m2 | 0.2 | 300 |
| m3 | 0.4 | 300 |
| m4 | 0.6 | 300 |
| m5 | 9 | 100 |
| m6 | 9.1 | 5000 |
| m7 | 9.2 | 100 |
Argument 2
3Argument 3
1000Argument 4
5Returns
[
{
'reason': 'rows',
'rows': [
'm1',
'm2',
'm3'
],
'bytes': 900,
'opened_at': 0,
'flushed_at': 0.6
},
{
'reason': 'time',
'rows': [
'm4'
],
'bytes': 300,
'opened_at': 0.6,
'flushed_at': 9
},
{
'reason': 'bytes',
'rows': [
'm5'
],
'bytes': 100,
'opened_at': 9,
'flushed_at': 9.1
},
{
'reason': 'bytes',
'rows': [
'm6'
],
'bytes': 5000,
'opened_at': 9.1,
'flushed_at': 9.2
},
{
'reason': 'end_of_stream',
'rows': [
'm7'
],
'bytes': 100,
'opened_at': 9.2,
'flushed_at': 9.2
}
]nothing arrives
An idle stream produces no flushes, in particular not one empty batch.
Input
Argument 1
[] (empty list)
Argument 2
3Argument 3
1000Argument 4
5Returns
[] (empty list)
one record
A single record still has to be flushed at the end of the stream rather than held forever.
Input
Argument 1
| id | at | bytes |
|---|---|---|
| solo | 1 | 10 |
Argument 2
3Argument 3
1000Argument 4
5Returns
[
{
'reason': 'end_of_stream',
'rows': [
'solo'
],
'bytes': 10,
'opened_at': 1,
'flushed_at': 1
}
]Constraints
max_rows rows, when the record's timestamp is more than max_delay after the batch opened, or when adding it would exceed max_bytes.rows, time or bytes. When more than one applies, the first is the reason.max_bytes goes out alone rather than blocking the stream.end_of_stream, timed at the last record.Worked example
With at most 3 rows, 1000 bytes and 5 seconds, the first four records arrive inside a second. The batch fills to three rows at 900 bytes and flushes for rows; m4 opens the next one at 0.6.
Then the stream goes quiet until 9.0. The batch holding m4 opened at 0.6, so by the time m5 arrives it is 8.4 seconds old — well past the 5-second delay — and flushes for time. Measuring the delay from the previous flush instead of from when the batch opened is the mistake that makes records sit in memory for unbounded periods: each new record resets the clock, and a slow trickle never triggers it at all.
m6 is 5000 bytes, larger than the whole budget. The byte rule guarded by 'the batch is not empty' lets it flush the batch holding m5 and then go out on its own, rather than being tested against an empty batch forever.
What this tests
Three interacting limits with a defined precedence, and the difference between batch age and idle time. The reason code is what makes the policy tunable, and it is the part that gets left out.
plan_flushes(records: list[dict], max_rows: int, max_bytes: int, max_delay: float) -> list[dict]Submit for review to find out what your query gets right, what it gets wrong, and how it compares with the best working query for this exercise.
This scenario runs a full workspace — editor, canvas and results side by side. It needs a laptop or desktop to be usable. Open this page on a bigger screen to start building.