Our current queue design was written for messages that contained a single record. In the world of RecordBatches, that leads to wildly different memory usage depending on the batch size.
This PR replaces our bounded channels with unbounded channels wrapped by a new pair of structs, BatchSender and BatchReceiver, which bound message sending based on the number of rows in each batch, rather than just the number of batches. This max queue size is also now configurable via the QUEUE_SIZE environment variable.
I've also added a number of new metrics to help users understand how queues contribute to memory usage.
--
This PR also includes a separate set of changes that fix issues that prevented startup of larger, networked pipelines. In particular, the startup process occurs synchronously within a call to "start_execution" from the controller. The controller gRPC client had a 30 second timeout for RPCs, so if this process took longer than 30 seconds it would fail. I was observing this commonly happening with large, networked pipelines.
I've addressed this in two parts:
Increasing the timeout to 90 seconds
Parallelizing the startup process by operator, where previously each operator was initialized independently
There is now a barrier that prevents operator from running until all operators are ready. This prevents chaotic behavior at the start of the pipeline, where a source will queue up data before other operators are ready to read it.
Our current queue design was written for messages that contained a single record. In the world of RecordBatches, that leads to wildly different memory usage depending on the batch size.
This PR replaces our bounded channels with unbounded channels wrapped by a new pair of structs, BatchSender and BatchReceiver, which bound message sending based on the number of rows in each batch, rather than just the number of batches. This max queue size is also now configurable via the QUEUE_SIZE environment variable.
I've also added a number of new metrics to help users understand how queues contribute to memory usage.
--
This PR also includes a separate set of changes that fix issues that prevented startup of larger, networked pipelines. In particular, the startup process occurs synchronously within a call to "start_execution" from the controller. The controller gRPC client had a 30 second timeout for RPCs, so if this process took longer than 30 seconds it would fail. I was observing this commonly happening with large, networked pipelines.
I've addressed this in two parts: