Revert "Temporarily disable all LIVE VIEW tests because this feature has subtle bugs that manifestate under TSan"

This reverts commit 96869d405f.

dbms/tests/queries/0_stateless/00960_live_view_watch_events_live.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+import os
+import sys
+import signal
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1, client(name='client2>', log=log) as client2:
+    client1.expect(prompt)
+    client2.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send(' DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+    client1.send('WATCH test.lv EVENTS')
+    client1.expect('1.*' + end_of_block)
+    client2.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+    client1.expect('2.*' + end_of_block)
+    client2.send('INSERT INTO test.mt VALUES (4),(5),(6)')
+    client1.expect('3.*' + end_of_block)
+    # send Ctrl-C
+    client1.send('\x03', eol='')
+    match = client1.expect('(%s)|([#\$] )' % prompt)
+    if match.groups()[1]:
+        client1.send(client1.command)
+        client1.expect(prompt)
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00961_temporary_live_view_watch.reference

@@ -0,0 +1,3 @@
+0	1
+6	2
+21	3

dbms/tests/queries/0_stateless/00961_temporary_live_view_watch.sql

@@ -0,0 +1,18 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE TEMPORARY LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt;
+
+WATCH test.lv LIMIT 0;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+WATCH test.lv LIMIT 0;
+
+INSERT INTO test.mt VALUES (4),(5),(6);
+
+WATCH test.lv LIMIT 0;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00962_temporary_live_view_watch_live.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+import os
+import sys
+import signal
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1, client(name='client2>', log=log) as client2:
+    client1.expect(prompt)
+    client2.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE TEMPORARY LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+    client1.send('WATCH test.lv')
+    client1.expect(r'0.*1' + end_of_block)
+    client2.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+    client1.expect(r'6.*2' + end_of_block)
+    client2.send('INSERT INTO test.mt VALUES (4),(5),(6)')
+    client1.expect(r'21.*3' + end_of_block)
+    # send Ctrl-C
+    client1.send('\x03', eol='')
+    match = client1.expect('(%s)|([#\$] )' % prompt)
+    if match.groups()[1]:
+        client1.send(client1.command)
+        client1.expect(prompt)
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00963_temporary_live_view_watch_live_timeout.py.disabled

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+import os
+import sys
+import signal
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1, client(name='client2>', log=log) as client2:
+    client1.expect(prompt)
+    client2.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('SET temporary_live_view_timeout=1')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE TEMPORARY LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+    client1.send('WATCH test.lv')
+    client1.expect(r'0.*1' + end_of_block)
+    client2.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+    client2.expect(prompt)
+    client1.expect(r'6.*2' + end_of_block)
+    client2.send('INSERT INTO test.mt VALUES (4),(5),(6)')
+    client2.expect(prompt)
+    client1.expect(r'21.*3' + end_of_block)
+    # send Ctrl-C
+    client1.send('\x03', eol='')
+    match = client1.expect('(%s)|([#\$] )' % prompt)
+    if match.groups()[1]:
+        client1.send(client1.command)
+        client1.expect(prompt)
+    client1.send('SELECT sleep(1)')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.lv')
+    client1.expect('Table test.lv doesn\'t exist')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00964_live_view_watch_events_heartbeat.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python
+import os
+import sys
+import signal
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1, client(name='client2>', log=log) as client2:
+    client1.expect(prompt)
+    client2.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send(' DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('SET live_view_heartbeat_interval=1')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE TEMPORARY LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+    client1.send('WATCH test.lv EVENTS')
+    client2.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+    client1.expect('2.*' + end_of_block)
+    client1.expect('Progress: 2.00 rows.*\)')
+    # wait for heartbeat
+    client1.expect('Progress: 2.00 rows.*\)')
+    # send Ctrl-C
+    client1.send('\x03', eol='')
+    match = client1.expect('(%s)|([#\$] )' % prompt)
+    if match.groups()[1]:
+        client1.send(client1.command)
+        client1.expect(prompt)    
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00965_live_view_watch_heartbeat.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python
+import os
+import sys
+import signal
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1, client(name='client2>', log=log) as client2:
+    client1.expect(prompt)
+    client2.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send(' DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('SET live_view_heartbeat_interval=1')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE TEMPORARY LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+    client1.send('WATCH test.lv')
+    client1.expect(r'0.*1' + end_of_block)
+    client2.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+    client1.expect(r'6.*2' + end_of_block)
+    client1.expect('Progress: 2.00 rows.*\)')
+    # wait for heartbeat
+    client1.expect('Progress: 2.00 rows.*\)')
+    # send Ctrl-C
+    client1.send('\x03', eol='')
+    match = client1.expect('(%s)|([#\$] )' % prompt)
+    if match.groups()[1]:
+        client1.send(client1.command)
+        client1.expect(prompt)    
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00966_live_view_watch_events_http.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python
+import os
+import sys
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+from httpclient import client as http_client
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1:
+    client1.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send(' DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+
+
+    with http_client({'method':'GET', 'url': '/?query=WATCH%20test.lv%20EVENTS'}, name='client2>', log=log) as client2:
+        client2.expect('.*1\n')
+        client1.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+        client1.expect(prompt)
+        client2.expect('.*2\n')
+
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00967_live_view_watch_http.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python
+import os
+import sys
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+from httpclient import client as http_client
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1:
+    client1.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send(' DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+
+
+    with http_client({'method':'GET', 'url':'/?query=WATCH%20test.lv'}, name='client2>', log=log) as client2:
+        client2.expect('.*0\t1\n')
+        client1.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+        client1.expect(prompt)
+        client2.expect('.*6\t2\n')
+
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00968_live_view_select_format_jsoneachrowwithprogress.reference

@@ -0,0 +1,4 @@
+{"row":{"a":1}}
+{"row":{"a":2}}
+{"row":{"a":3}}
+{"progress":{"read_rows":"3","read_bytes":"36","written_rows":"0","written_bytes":"0","total_rows_to_read":"0"}}

dbms/tests/queries/0_stateless/00968_live_view_select_format_jsoneachrowwithprogress.sql

@@ -0,0 +1,12 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE LIVE VIEW test.lv AS SELECT * FROM test.mt;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+SELECT * FROM test.lv FORMAT JSONEachRowWithProgress;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00969_live_view_watch_format_jsoneachrowwithprogress.reference

@@ -0,0 +1,6 @@
+{"row":{"sum(a)":"0","_version":"1"}}
+{"progress":{"read_rows":"1","read_bytes":"16","written_rows":"0","written_bytes":"0","total_rows_to_read":"0"}}
+{"row":{"sum(a)":"6","_version":"2"}}
+{"progress":{"read_rows":"1","read_bytes":"16","written_rows":"0","written_bytes":"0","total_rows_to_read":"0"}}
+{"row":{"sum(a)":"21","_version":"3"}}
+{"progress":{"read_rows":"1","read_bytes":"16","written_rows":"0","written_bytes":"0","total_rows_to_read":"0"}}

dbms/tests/queries/0_stateless/00969_live_view_watch_format_jsoneachrowwithprogress.sql

@@ -0,0 +1,18 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt;
+
+WATCH test.lv LIMIT 0 FORMAT JSONEachRowWithProgress;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+WATCH test.lv LIMIT 0 FORMAT JSONEachRowWithProgress;
+
+INSERT INTO test.mt VALUES (4),(5),(6);
+
+WATCH test.lv LIMIT 0 FORMAT JSONEachRowWithProgress;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00970_live_view_watch_events_http_heartbeat.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python
+import os
+import sys
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+from httpclient import client as http_client
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1:
+    client1.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send(' DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+
+
+    with http_client({'method':'GET', 'url': '/?live_view_heartbeat_interval=1&query=WATCH%20test.lv%20EVENTS%20FORMAT%20JSONEachRowWithProgress'}, name='client2>', log=log) as client2:
+        client2.expect('{"progress":{"read_rows":"1","read_bytes":"8","written_rows":"0","written_bytes":"0","total_rows_to_read":"0"}}\n', escape=True)
+        client2.expect('{"row":{"version":"1"}', escape=True)
+        client2.expect('{"progress":{"read_rows":"1","read_bytes":"8","written_rows":"0","written_bytes":"0","total_rows_to_read":"0"}}', escape=True)
+        # heartbeat is provided by progress message
+        client2.expect('{"progress":{"read_rows":"1","read_bytes":"8","written_rows":"0","written_bytes":"0","total_rows_to_read":"0"}}', escape=True)
+
+        client1.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+        client1.expect(prompt)
+
+        client2.expect('{"row":{"version":"2"}}\n', escape=True)
+
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00971_live_view_watch_http_heartbeat.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python
+import os
+import sys
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+from httpclient import client as http_client
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1:
+    client1.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send(' DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+
+    with http_client({'method':'GET', 'url':'/?live_view_heartbeat_interval=1&query=WATCH%20test.lv%20FORMAT%20JSONEachRowWithProgress'}, name='client2>', log=log) as client2:
+        client2.expect('"progress".*',)
+        client2.expect('{"row":{"sum(a)":"0","_version":"1"}}\n', escape=True)
+        client2.expect('"progress".*\n')
+        # heartbeat is provided by progress message
+        client2.expect('"progress".*\n')
+
+        client1.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+        client1.expect(prompt)
+
+        client2.expect('"progress".*"read_rows":"2".*\n')
+        client2.expect('{"row":{"sum(a)":"6","_version":"2"}}\n', escape=True)
+
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00972_live_view_select_1.reference

@@ -0,0 +1 @@
+1

dbms/tests/queries/0_stateless/00972_live_view_select_1.sql

@@ -0,0 +1,7 @@
+DROP TABLE IF EXISTS test.lv;
+
+CREATE LIVE VIEW test.lv AS SELECT 1;
+
+SELECT * FROM test.lv;
+
+DROP TABLE test.lv;

dbms/tests/queries/0_stateless/00973_live_view_select.reference

@@ -0,0 +1,4 @@
+6	1
+6	1
+12	2
+12	2

dbms/tests/queries/0_stateless/00973_live_view_select.sql

@@ -0,0 +1,18 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+SELECT *,_version FROM test.lv;
+SELECT *,_version FROM test.lv;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+SELECT *,_version FROM test.lv;
+SELECT *,_version FROM test.lv;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00974_live_view_select_with_aggregation.reference

@@ -0,0 +1,2 @@
+6
+21

dbms/tests/queries/0_stateless/00974_live_view_select_with_aggregation.sql

@@ -0,0 +1,16 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE LIVE VIEW test.lv AS SELECT * FROM test.mt;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+SELECT sum(a) FROM test.lv;
+
+INSERT INTO test.mt VALUES (4),(5),(6);
+
+SELECT sum(a) FROM test.lv;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00975_live_view_create.sql

@@ -0,0 +1,7 @@
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE LIVE VIEW test.lv AS SELECT * FROM test.mt;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00976_live_view_select_version.reference

@@ -0,0 +1,3 @@
+1	1
+2	1
+3	1

dbms/tests/queries/0_stateless/00976_live_view_select_version.sql

@@ -0,0 +1,12 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE LIVE VIEW test.lv AS SELECT * FROM test.mt;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+SELECT *,_version FROM test.lv;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00977_live_view_watch_events.reference

@@ -0,0 +1,3 @@
+1
+2
+3

dbms/tests/queries/0_stateless/00977_live_view_watch_events.sql

@@ -0,0 +1,18 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt;
+
+WATCH test.lv EVENTS LIMIT 0;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+WATCH test.lv EVENTS LIMIT 0;
+
+INSERT INTO test.mt VALUES (4),(5),(6);
+
+WATCH test.lv EVENTS LIMIT 0;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00978_live_view_watch.reference

@@ -0,0 +1,3 @@
+0	1
+6	2
+21	3

dbms/tests/queries/0_stateless/00978_live_view_watch.sql

@@ -0,0 +1,18 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt;
+
+WATCH test.lv LIMIT 0;
+
+INSERT INTO test.mt VALUES (1),(2),(3);
+
+WATCH test.lv LIMIT 0;
+
+INSERT INTO test.mt VALUES (4),(5),(6);
+
+WATCH test.lv LIMIT 0;
+
+DROP TABLE test.lv;
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00979_live_view_watch_live.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python
+import os
+import sys
+import signal
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.join(CURDIR, 'helpers'))
+
+from client import client, prompt, end_of_block
+
+log = None
+# uncomment the line below for debugging
+#log=sys.stdout
+
+with client(name='client1>', log=log) as client1, client(name='client2>', log=log) as client2:
+    client1.expect(prompt)
+    client2.expect(prompt)
+
+    client1.send('DROP TABLE IF EXISTS test.lv')
+    client1.expect(prompt)
+    client1.send(' DROP TABLE IF EXISTS test.mt')
+    client1.expect(prompt)
+    client1.send('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()')
+    client1.expect(prompt)
+    client1.send('CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt')
+    client1.expect(prompt)
+    client1.send('WATCH test.lv')
+    client1.expect(r'0.*1' + end_of_block)
+    client2.send('INSERT INTO test.mt VALUES (1),(2),(3)')
+    client1.expect(r'6.*2' + end_of_block)
+    client2.expect(prompt)
+    client2.send('INSERT INTO test.mt VALUES (4),(5),(6)')
+    client1.expect(r'21.*3' + end_of_block)
+    client2.expect(prompt)
+    for i in range(1,129):
+       client2.send('INSERT INTO test.mt VALUES (1)')
+       client1.expect(r'%d.*%d' % (21+i, 3+i) + end_of_block)
+       client2.expect(prompt)
+    # send Ctrl-C
+    client1.send('\x03', eol='')
+    match = client1.expect('(%s)|([#\$] )' % prompt)
+    if match.groups()[1]:
+        client1.send(client1.command)
+        client1.expect(prompt)    
+    client1.send('DROP TABLE test.lv')
+    client1.expect(prompt)
+    client1.send('DROP TABLE test.mt')
+    client1.expect(prompt)

dbms/tests/queries/0_stateless/00980_create_temporary_live_view.reference

@@ -0,0 +1,3 @@
+temporary_live_view_timeout	5
+live_view_heartbeat_interval	15
+0

dbms/tests/queries/0_stateless/00980_create_temporary_live_view.sql

@@ -0,0 +1,15 @@
+DROP TABLE IF EXISTS test.lv;
+DROP TABLE IF EXISTS test.mt;
+
+SELECT name, value from system.settings WHERE name = 'temporary_live_view_timeout';
+SELECT name, value from system.settings WHERE name = 'live_view_heartbeat_interval';
+
+SET temporary_live_view_timeout=1;
+CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple();
+CREATE TEMPORARY LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt;
+
+SHOW TABLES LIKE 'lv';
+SELECT sleep(2);
+SHOW TABLES LIKE 'lv';
+
+DROP TABLE test.mt;

dbms/tests/queries/0_stateless/00991_live_view_watch_event_live.python

@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+
+import subprocess
+import threading
+import Queue as queue
+import os
+import sys
+import signal
+
+
+CLICKHOUSE_CLIENT = os.environ.get('CLICKHOUSE_CLIENT')
+CLICKHOUSE_CURL = os.environ.get('CLICKHOUSE_CURL')
+CLICKHOUSE_URL = os.environ.get('CLICKHOUSE_URL')
+
+
+def send_query(query):
+    cmd = list(CLICKHOUSE_CLIENT.split())
+    cmd += ['--query', query]
+    # print(cmd)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT).stdout
+
+
+def send_query_in_process_group(query):
+    cmd = list(CLICKHOUSE_CLIENT.split())
+    cmd += ['--query', query]
+    # print(cmd)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, preexec_fn=os.setsid)
+
+
+def read_lines_and_push_to_queue(pipe, queue):
+    try:
+        for line in iter(pipe.readline, ''):
+            line = line.strip()
+            print(line)
+            sys.stdout.flush()
+            queue.put(line)
+    except KeyboardInterrupt:
+        pass
+
+    queue.put(None)
+
+
+def test():
+    send_query('DROP TABLE IF EXISTS test.lv').read()
+    send_query('DROP TABLE IF EXISTS test.mt').read()
+    send_query('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()').read()
+    send_query('CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt').read()
+
+    q = queue.Queue()
+    p = send_query_in_process_group('WATCH test.lv')
+    thread = threading.Thread(target=read_lines_and_push_to_queue, args=(p.stdout, q))
+    thread.start()
+
+    line = q.get()
+    print(line)
+    assert (line == '0\t1')
+
+    send_query('INSERT INTO test.mt VALUES (1),(2),(3)').read()
+    line = q.get()
+    print(line)
+    assert (line == '6\t2')
+
+    send_query('INSERT INTO test.mt VALUES (4),(5),(6)').read()
+    line = q.get()
+    print(line)
+    assert (line == '21\t3')
+
+    # Send Ctrl+C to client.
+    os.killpg(os.getpgid(p.pid), signal.SIGINT)
+    # This insert shouldn't affect lv.
+    send_query('INSERT INTO test.mt VALUES (7),(8),(9)').read()
+    line = q.get()
+    print(line)
+    assert (line is None)
+
+    send_query('DROP TABLE if exists test.lv').read()
+    send_query('DROP TABLE if exists test.lv').read()
+
+    thread.join()
+
+test()

dbms/tests/queries/0_stateless/00991_live_view_watch_event_live.reference

@@ -0,0 +1,7 @@
+0	1
+0	1
+6	2
+6	2
+21	3
+21	3
+None

dbms/tests/queries/0_stateless/00991_live_view_watch_event_live.sh.disabled

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+python $CURDIR/00991_live_view_watch_event_live.python

dbms/tests/queries/0_stateless/00991_live_view_watch_http.python

@@ -0,0 +1,63 @@
+#!/usr/bin/env python
+
+import subprocess
+import threading
+import Queue as queue
+import os
+import sys
+
+
+CLICKHOUSE_CLIENT = os.environ.get('CLICKHOUSE_CLIENT')
+CLICKHOUSE_CURL = os.environ.get('CLICKHOUSE_CURL')
+CLICKHOUSE_URL = os.environ.get('CLICKHOUSE_URL')
+
+
+def send_query(query):
+    cmd = list(CLICKHOUSE_CLIENT.split())
+    cmd += ['--query', query]
+    # print(cmd)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT).stdout
+
+
+def send_http_query(query):
+    cmd =  list(CLICKHOUSE_CURL.split()) # list(['curl', '-sSN', '--max-time', '10'])
+    cmd += ['-sSN', CLICKHOUSE_URL, '-d', query]
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT).stdout
+
+
+def read_lines_and_push_to_queue(pipe, queue):
+    for line in iter(pipe.readline, ''):
+        line = line.strip()
+        print(line)
+        sys.stdout.flush()
+        queue.put(line)
+
+    queue.put(None)
+
+
+def test():
+    send_query('DROP TABLE IF EXISTS test.lv').read()
+    send_query('DROP TABLE IF EXISTS test.mt').read()
+    send_query('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()').read()
+    send_query('CREATE LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt').read()
+
+    q = queue.Queue()
+    pipe = send_http_query('WATCH test.lv')
+    thread = threading.Thread(target=read_lines_and_push_to_queue, args=(pipe, q))
+    thread.start()
+
+    line = q.get()
+    print(line)
+    assert (line == '0\t1')
+
+    send_query('INSERT INTO test.mt VALUES (1),(2),(3)').read()
+    line = q.get()
+    print(line)
+    assert (line == '6\t2')
+
+    send_query('DROP TABLE if exists test.lv').read()
+    send_query('DROP TABLE if exists test.lv').read()
+
+    thread.join()
+
+test()

dbms/tests/queries/0_stateless/00991_live_view_watch_http.reference

@@ -0,0 +1,4 @@
+0	1
+0	1
+6	2
+6	2

dbms/tests/queries/0_stateless/00991_live_view_watch_http.sh.disabled

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+python $CURDIR/00991_live_view_watch_http.python

dbms/tests/queries/0_stateless/00991_temporary_live_view_watch_events_heartbeat.python

@@ -0,0 +1,83 @@
+#!/usr/bin/env python
+
+import subprocess
+import threading
+import Queue as queue
+import os
+import sys
+import signal
+
+
+CLICKHOUSE_CLIENT = os.environ.get('CLICKHOUSE_CLIENT')
+CLICKHOUSE_CURL = os.environ.get('CLICKHOUSE_CURL')
+CLICKHOUSE_URL = os.environ.get('CLICKHOUSE_URL')
+
+
+def send_query(query):
+    cmd = list(CLICKHOUSE_CLIENT.split())
+    cmd += ['--query', query]
+    # print(cmd)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT).stdout
+
+
+def send_query_in_process_group(query):
+    cmd = list(CLICKHOUSE_CLIENT.split())
+    cmd += ['--query', query, '--live_view_heartbeat_interval=1', '--progress']
+    # print(cmd)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, preexec_fn=os.setsid)
+
+
+def read_lines_and_push_to_queue(pipe, queue):
+    try:
+        for line in iter(pipe.readline, ''):
+            line = line.strip()
+            # print(line)
+            sys.stdout.flush()
+            queue.put(line)
+    except KeyboardInterrupt:
+        pass
+
+    queue.put(None)
+
+
+def test():
+    send_query('DROP TABLE IF EXISTS test.lv').read()
+    send_query('DROP TABLE IF EXISTS test.mt').read()
+    send_query('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()').read()
+    send_query('CREATE TEMPORARY LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt').read()
+
+    q = queue.Queue()
+    p = send_query_in_process_group('WATCH test.lv')
+    thread = threading.Thread(target=read_lines_and_push_to_queue, args=(p.stdout, q))
+    thread.start()
+
+    line = q.get()
+    # print(line)
+    assert (line.endswith('0\t1'))
+    assert ('Progress: 0.00 rows' in line)
+
+    send_query('INSERT INTO test.mt VALUES (1),(2),(3)').read()
+    line = q.get()
+    assert (line.endswith('6\t2'))
+    assert ('Progress: 1.00 rows' in line)
+
+    # send_query('INSERT INTO test.mt VALUES (4),(5),(6)').read()
+    # line = q.get()
+    # print(line)
+    # assert (line.endswith('6\t2'))
+    # assert ('Progress: 1.00 rows' in line)
+
+    # Send Ctrl+C to client.
+    os.killpg(os.getpgid(p.pid), signal.SIGINT)
+    # This insert shouldn't affect lv.
+    send_query('INSERT INTO test.mt VALUES (7),(8),(9)').read()
+    line = q.get()
+    # print(line)
+    # assert (line is None)
+
+    send_query('DROP TABLE if exists test.lv').read()
+    send_query('DROP TABLE if exists test.lv').read()
+
+    thread.join()
+
+test()

dbms/tests/queries/0_stateless/00991_temporary_live_view_watch_events_heartbeat.sh.disabled

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+python $CURDIR/00991_temporary_live_view_watch_events_heartbeat.python

dbms/tests/queries/0_stateless/00991_temporary_live_view_watch_live.python

@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+
+import subprocess
+import threading
+import Queue as queue
+import os
+import sys
+import signal
+
+
+CLICKHOUSE_CLIENT = os.environ.get('CLICKHOUSE_CLIENT')
+CLICKHOUSE_CURL = os.environ.get('CLICKHOUSE_CURL')
+CLICKHOUSE_URL = os.environ.get('CLICKHOUSE_URL')
+
+
+def send_query(query):
+    cmd = list(CLICKHOUSE_CLIENT.split())
+    cmd += ['--query', query]
+    # print(cmd)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT).stdout
+
+
+def send_query_in_process_group(query):
+    cmd = list(CLICKHOUSE_CLIENT.split())
+    cmd += ['--query', query]
+    # print(cmd)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, preexec_fn=os.setsid)
+
+
+def read_lines_and_push_to_queue(pipe, queue):
+    try:
+        for line in iter(pipe.readline, ''):
+            line = line.strip()
+            print(line)
+            sys.stdout.flush()
+            queue.put(line)
+    except KeyboardInterrupt:
+        pass
+
+    queue.put(None)
+
+
+def test():
+    send_query('DROP TABLE IF EXISTS test.lv').read()
+    send_query('DROP TABLE IF EXISTS test.mt').read()
+    send_query('CREATE TABLE test.mt (a Int32) Engine=MergeTree order by tuple()').read()
+    send_query('CREATE TEMPORARY LIVE VIEW test.lv AS SELECT sum(a) FROM test.mt').read()
+
+    q = queue.Queue()
+    p = send_query_in_process_group('WATCH test.lv')
+    thread = threading.Thread(target=read_lines_and_push_to_queue, args=(p.stdout, q))
+    thread.start()
+
+    line = q.get()
+    print(line)
+    assert (line == '0\t1')
+
+    send_query('INSERT INTO test.mt VALUES (1),(2),(3)').read()
+    line = q.get()
+    print(line)
+    assert (line == '6\t2')
+
+    send_query('INSERT INTO test.mt VALUES (4),(5),(6)').read()
+    line = q.get()
+    print(line)
+    assert (line == '21\t3')
+
+    # Send Ctrl+C to client.
+    os.killpg(os.getpgid(p.pid), signal.SIGINT)
+    # This insert shouldn't affect lv.
+    send_query('INSERT INTO test.mt VALUES (7),(8),(9)').read()
+    line = q.get()
+    print(line)
+    assert (line is None)
+
+    send_query('DROP TABLE if exists test.lv').read()
+    send_query('DROP TABLE if exists test.lv').read()
+
+    thread.join()
+
+test()

dbms/tests/queries/0_stateless/00991_temporary_live_view_watch_live.reference

@@ -0,0 +1,7 @@
+0	1
+0	1
+6	2
+6	2
+21	3
+21	3
+None

dbms/tests/queries/0_stateless/00991_temporary_live_view_watch_live.sh.disabled

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+python $CURDIR/00991_temporary_live_view_watch_live.python

dbms/tests/queries/0_stateless/helpers/client.py

@@ -0,0 +1,36 @@
+import os
+import sys
+import time
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+
+sys.path.insert(0, os.path.join(CURDIR))
+
+import uexpect
+
+prompt = ':\) '
+end_of_block = r'.*\r\n.*\r\n'
+
+class client(object):
+    def __init__(self, command=None, name='', log=None):
+        self.client = uexpect.spawn(['/bin/bash','--noediting'])
+        if command is None:
+            command = os.environ.get('CLICKHOUSE_BINARY', 'clickhouse') + '-client'
+        self.client.command = command
+        self.client.eol('\r')
+        self.client.logger(log, prefix=name)
+        self.client.timeout(20)
+        self.client.expect('[#\$] ', timeout=2)
+        self.client.send(command)
+
+    def __enter__(self):
+        return self.client.__enter__()
+
+    def __exit__(self, type, value, traceback):
+        self.client.reader['kill_event'].set()
+        # send Ctrl-C
+        self.client.send('\x03', eol='')
+        time.sleep(0.3)
+        self.client.send('quit', eol='\r')
+        self.client.send('\x03', eol='')
+        return self.client.__exit__(type, value, traceback)

dbms/tests/queries/0_stateless/helpers/httpclient.py

@@ -0,0 +1,14 @@
+import os
+import sys
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+
+sys.path.insert(0, os.path.join(CURDIR))
+
+import httpexpect
+
+def client(request, name='', log=None):
+    client = httpexpect.spawn({'host':'localhost','port':8123}, request)
+    client.logger(log, prefix=name)
+    client.timeout(20)
+    return client

dbms/tests/queries/0_stateless/helpers/httpexpect.py

@@ -0,0 +1,73 @@
+# Copyright (c) 2019 Vitaliy Zakaznikov
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import os
+import sys
+import httplib
+
+CURDIR = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, CURDIR)
+
+import uexpect
+
+from threading import Thread, Event
+from Queue import Queue, Empty
+
+class IO(uexpect.IO):
+    def __init__(self, connection, response, queue, reader):
+        self.connection = connection
+        self.response = response
+        super(IO, self).__init__(None, None, queue, reader)
+
+    def write(self, data):
+        raise NotImplementedError
+
+    def close(self, force=True):
+        self.reader['kill_event'].set()
+        self.connection.close()
+        if self._logger:
+            self._logger.write('\n')
+            self._logger.flush()
+
+
+def reader(response, queue, kill_event):
+    while True:
+        try:
+            if kill_event.is_set():
+                break
+            data = response.read(1)
+            queue.put(data)
+        except Exception, e:
+            if kill_event.is_set():
+                break
+            raise
+
+def spawn(connection, request):
+    connection = httplib.HTTPConnection(**connection)
+    connection.request(**request)
+    response = connection.getresponse()
+
+    queue = Queue()
+    reader_kill_event = Event()
+    thread = Thread(target=reader, args=(response, queue, reader_kill_event))
+    thread.daemon = True
+    thread.start()
+
+    return IO(connection, response, queue, reader={'thread':thread, 'kill_event':reader_kill_event})
+
+if __name__ == '__main__':
+    with http({'host':'localhost','port':8123},{'method':'GET', 'url':'?query=SELECT%201'}) as client:
+        client.logger(sys.stdout)
+        client.timeout(2)
+        print client.response.status, client.response.reason
+        client.expect('1\n')

dbms/tests/queries/0_stateless/helpers/uexpect.py

@@ -0,0 +1,206 @@
+# Copyright (c) 2019 Vitaliy Zakaznikov
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import os
+import pty
+import time
+import sys
+import re
+
+from threading import Thread, Event
+from subprocess import Popen
+from Queue import Queue, Empty
+
+class TimeoutError(Exception):
+    def __init__(self, timeout):
+        self.timeout = timeout
+
+    def __str__(self):
+        return 'Timeout %.3fs' % float(self.timeout)
+
+class ExpectTimeoutError(Exception):
+    def __init__(self, pattern, timeout, buffer):
+        self.pattern = pattern
+        self.timeout = timeout
+        self.buffer = buffer
+
+    def __str__(self):
+        s = 'Timeout %.3fs ' % float(self.timeout)
+        if self.pattern:
+            s += 'for %s ' % repr(self.pattern.pattern)
+        if self.buffer:
+            s += 'buffer %s ' % repr(self.buffer[:])
+            s += 'or \'%s\'' % ','.join(['%x' % ord(c) for c in self.buffer[:]])
+        return s
+
+class IO(object):
+    class EOF(object):
+        pass
+
+    class Timeout(object):
+        pass
+
+    EOF = EOF
+    TIMEOUT = Timeout
+
+    class Logger(object):
+        def __init__(self, logger, prefix=''):
+            self._logger = logger
+            self._prefix = prefix
+
+        def write(self, data):
+            self._logger.write(('\n' + data).replace('\n','\n' + self._prefix))
+
+        def flush(self):
+            self._logger.flush()
+
+    def __init__(self, process, master, queue, reader):
+        self.process = process
+        self.master = master
+        self.queue = queue
+        self.buffer = None
+        self.before = None
+        self.after = None
+        self.match = None
+        self.pattern = None
+        self.reader = reader
+        self._timeout = None
+        self._logger = None
+        self._eol = ''
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.close()
+
+    def logger(self, logger=None, prefix=''):
+        if logger:
+            self._logger = self.Logger(logger, prefix=prefix)
+        return self._logger
+
+    def timeout(self, timeout=None):
+        if timeout:
+            self._timeout = timeout
+        return self._timeout
+
+    def eol(self, eol=None):
+        if eol:
+            self._eol = eol
+        return self._eol
+
+    def close(self, force=True):
+        self.reader['kill_event'].set()
+        os.system('pkill -TERM -P %d' % self.process.pid)
+        if force:
+            self.process.kill()
+        else:
+            self.process.terminate()
+        os.close(self.master)
+        if self._logger:
+            self._logger.write('\n')
+            self._logger.flush()
+
+    def send(self, data, eol=None):
+        if eol is None:
+            eol = self._eol
+        return self.write(data + eol)
+
+    def write(self, data):
+        return os.write(self.master, data)
+
+    def expect(self, pattern, timeout=None, escape=False):
+        self.match = None
+        self.before = None
+        self.after = None
+        if escape:
+            pattern = re.escape(pattern)
+        pattern = re.compile(pattern)
+        if timeout is None:
+            timeout = self._timeout
+        timeleft = timeout
+        while True:
+            start_time = time.time()
+            if self.buffer is not None:
+                self.match = pattern.search(self.buffer, 0)
+                if self.match is not None:
+                    self.after = self.buffer[self.match.start():self.match.end()]
+                    self.before = self.buffer[:self.match.start()]
+                    self.buffer = self.buffer[self.match.end():]
+                    break
+            if timeleft < 0:
+                break
+            try:
+                data = self.read(timeout=timeleft, raise_exception=True)
+            except TimeoutError:
+                if self._logger:
+                    self._logger.write((self.buffer or '') + '\n')
+                    self._logger.flush()
+                exception = ExpectTimeoutError(pattern, timeout, self.buffer)
+                self.buffer = None
+                raise exception
+            timeleft -= (time.time() - start_time)
+            if data:
+                self.buffer = (self.buffer + data) if self.buffer else data
+        if self._logger:
+            self._logger.write((self.before or '') + (self.after or ''))
+            self._logger.flush()
+        if self.match is None:
+            exception = ExpectTimeoutError(pattern, timeout, self.buffer)
+            self.buffer = None
+            raise exception
+        return self.match
+
+    def read(self, timeout=0, raise_exception=False):
+        data = ''
+        timeleft = timeout
+        try:
+            while timeleft >= 0 :
+                start_time = time.time()
+                data += self.queue.get(timeout=timeleft)
+                if data:
+                    break
+                timeleft -= (time.time() - start_time)
+        except Empty:
+            if data:
+                return data
+            if raise_exception:
+                raise TimeoutError(timeout)
+            pass
+        if not data and raise_exception:
+            raise TimeoutError(timeout)
+
+        return data
+
+def spawn(command):
+    master, slave = pty.openpty()
+    process = Popen(command, preexec_fn=os.setsid, stdout=slave, stdin=slave, stderr=slave, bufsize=1)
+    os.close(slave)
+
+    queue = Queue()
+    reader_kill_event = Event()
+    thread = Thread(target=reader, args=(process, master, queue, reader_kill_event))
+    thread.daemon = True
+    thread.start()
+
+    return IO(process, master, queue, reader={'thread':thread, 'kill_event':reader_kill_event})
+
+def reader(process, out, queue, kill_event):
+    while True:
+        try:
+            data = os.read(out, 65536)
+            queue.put(data)
+        except:
+            if kill_event.is_set():
+                break
+            raise