Playing with various formats for the `executable()` table function got to a peculiar perf problem.
Query:
```sql
caspiandb6 :) SELECT * FROM executable('arrow_sum.py', 'ArrowStream', 'sum UInt64', (SELECT * FROM numbers(10*1000000)));

SELECT *
FROM executable('arrow_sum.py', 'ArrowStream', 'sum UInt64', (
    SELECT *
    FROM numbers(10 * 1000000)
))

Query id: b73e5936-22d1-4d4b-826e-18d3aab80e47

┌────────────sum─┐
│ 49999995000000 │
└────────────────┘

1 row in set. Elapsed: 1.131 sec. 

```
1.131 sec ? seems slow as hell
Let's decompose the query:
```bash
$ time ./clickhouse-client --query="SELECT * FROM numbers(10*1000000) FORMAT ArrowStream" > /tmp/file

real	0m0.097s
user	0m0.019s
sys	0m0.065s

$ time cat /tmp/file | ./user_scripts/arrow_sum.py >/dev/null

real	0m0.128s
user	0m0.182s
sys	0m0.301s
```
0.097 + 0.128 = 0.225 sec
This looks about right.

Let's try different array sizes:
```sql
caspiandb6 :) SELECT * FROM executable('arrow_sum.py', 'ArrowStream', 'sum UInt64', (SELECT * FROM numbers(1*1000000)));

SELECT *
FROM executable('arrow_sum.py', 'ArrowStream', 'sum UInt64', (
    SELECT *
    FROM numbers(1 * 1000000)
))

Query id: 26965ad9-8115-4b59-a0cd-c062a7e0658a

┌──────────sum─┐
│ 499999500000 │
└──────────────┘

1 row in set. Elapsed: 1.094 sec. 

caspiandb6 :) SELECT * FROM executable('arrow_sum.py', 'ArrowStream', 'sum UInt64', (SELECT * FROM numbers(100*1000000)));

SELECT *
FROM executable('arrow_sum.py', 'ArrowStream', 'sum UInt64', (
    SELECT *
    FROM numbers(100 * 1000000)
))

Query id: e3dea8a3-1fb6-4d25-9182-1d9aadd12e7b

┌──────────────sum─┐
│ 4999999950000000 │
└──────────────────┘

1 row in set. Elapsed: 1.302 sec. 
```
It looks like 1 sec is just added to all queries.
Seems like a famous `sleep 1` problem somewhere in the code, frankly. :)
@kitaisreal any idea what could be the problem?
I will continue to poke it meanwhile.

For the sake of completeness:
`arrow_sum.py`
```python
#!/usr/bin/python3

import sys
import traceback
import pyarrow as pa
import pyarrow.compute as pc

if __name__ == "__main__":
    fd = open("/tmp/debug.log", "w")
    sum = pa.scalar(0, type="UInt64")
    try:
        with pa.ipc.open_stream(sys.stdin.buffer) as reader:
            for b in reader:
                sum = pc.add(sum, pc.sum(b[0]))
        array = pa.array([sum.as_py()], type="UInt64")
        out_batch = pa.record_batch([array], names=['sum'])
        with pa.ipc.new_stream(sys.stdout.buffer, out_batch.schema) as writer:
            writer.write(out_batch)
    except Exception as e:
        traceback.print_exc(file=fd)
        fd.close()
    sys.stdout.flush()

```
{
    int status = 0;
    LOG_TRACE(getLogger(), "Try wait for shell command pid {} with timeout {}", pid, timeout_in_seconds);
    wait_called = true;
    struct timespec interval {.tv_sec = 1, .tv_nsec = 0};
    in.close();
    out.close();
    err.close();
    if (timeout_in_seconds == 0)
    {
        
        
        int waitpid_res = waitpid(pid, &status, WNOHANG);
        bool process_terminated_normally = (waitpid_res == pid);
        return process_terminated_normally;
    }
    
    
    while (timeout_in_seconds != 0)
    {
        int waitpid_res = waitpid(pid, &status, WNOHANG);
        bool process_terminated_normally = (waitpid_res == pid);
        if (process_terminated_normally)
        {