https://pastila.nl/?001ae728/52bc538966a011139c5dac06472b22f6
{
    if (ports_data.size() != 1)
        throw Exception("prepare without arguments is not supported for multi-port OffsetTransform",
                        ErrorCodes::LOGICAL_ERROR);
    return prepare({0}, {0});
}
OffsetTransform::Status OffsetTransform::preparePair(PortsData & data)
{
    auto & output = *data.output_port;
    auto & input = *data.input_port;
    
    bool output_finished = false;
    if (output.isFinished())
    {
        output_finished = true;
    }
    if (!output_finished && !output.canPush())
    {
        input.setNotNeeded();
        return Status::PortFull;
    }
    
    if (input.isFinished())
    {