Merge branch 'master' of github.com:yandex/ClickHouse

dbms/programs/client/ConnectionParameters.h

@@ -76,7 +76,8 @@ struct ConnectionParameters
         timeouts = ConnectionTimeouts(
             Poco::Timespan(config.getInt("connect_timeout", DBMS_DEFAULT_CONNECT_TIMEOUT_SEC), 0),
             Poco::Timespan(config.getInt("receive_timeout", DBMS_DEFAULT_RECEIVE_TIMEOUT_SEC), 0),
-            Poco::Timespan(config.getInt("send_timeout", DBMS_DEFAULT_SEND_TIMEOUT_SEC), 0));
+            Poco::Timespan(config.getInt("send_timeout", DBMS_DEFAULT_SEND_TIMEOUT_SEC), 0),
+            Poco::Timespan(config.getInt("tcp_keep_alive_timeout", 0), 0));
     }
 };
 

dbms/src/Client/Connection.cpp

@@ -77,6 +77,11 @@ void Connection::connect()
         socket->setReceiveTimeout(timeouts.receive_timeout);
         socket->setSendTimeout(timeouts.send_timeout);
         socket->setNoDelay(true);
+        if (timeouts.tcp_keep_alive_timeout.totalSeconds())
+        {
+            socket->setKeepAlive(true);
+            socket->setOption(SOL_TCP, TCP_KEEPIDLE, timeouts.tcp_keep_alive_timeout);
+        }
 
         in = std::make_shared<ReadBufferFromPocoSocket>(*socket);
         out = std::make_shared<WriteBufferFromPocoSocket>(*socket);

dbms/src/IO/ConnectionTimeouts.h

@@ -11,6 +11,7 @@ struct ConnectionTimeouts
     Poco::Timespan connection_timeout;
     Poco::Timespan send_timeout;
     Poco::Timespan receive_timeout;
+    Poco::Timespan tcp_keep_alive_timeout;
 
     ConnectionTimeouts() = default;
 
@@ -19,7 +20,19 @@ struct ConnectionTimeouts
                        const Poco::Timespan & receive_timeout_)
     : connection_timeout(connection_timeout_),
       send_timeout(send_timeout_),
-      receive_timeout(receive_timeout_)
+      receive_timeout(receive_timeout_),
+      tcp_keep_alive_timeout(0)
+    {
+    }
+
+    ConnectionTimeouts(const Poco::Timespan & connection_timeout_,
+                       const Poco::Timespan & send_timeout_,
+                       const Poco::Timespan & receive_timeout_,
+                       const Poco::Timespan & tcp_keep_alive_timeout_)
+    : connection_timeout(connection_timeout_),
+      send_timeout(send_timeout_),
+      receive_timeout(receive_timeout_),
+      tcp_keep_alive_timeout(tcp_keep_alive_timeout_)
     {
     }
 
@@ -35,19 +48,20 @@ struct ConnectionTimeouts
     {
         return ConnectionTimeouts(saturate(connection_timeout, limit),
                                   saturate(send_timeout, limit),
-                                  saturate(receive_timeout, limit));
+                                  saturate(receive_timeout, limit),
+                                  saturate(tcp_keep_alive_timeout, limit));
     }
 
     /// Timeouts for the case when we have just single attempt to connect.
     static ConnectionTimeouts getTCPTimeoutsWithoutFailover(const Settings & settings)
     {
-        return ConnectionTimeouts(settings.connect_timeout, settings.send_timeout, settings.receive_timeout);
+        return ConnectionTimeouts(settings.connect_timeout, settings.send_timeout, settings.receive_timeout, settings.tcp_keep_alive_timeout);
     }
 
     /// Timeouts for the case when we will try many addresses in a loop.
     static ConnectionTimeouts getTCPTimeoutsWithFailover(const Settings & settings)
     {
-        return ConnectionTimeouts(settings.connect_timeout_with_failover_ms, settings.send_timeout, settings.receive_timeout);
+        return ConnectionTimeouts(settings.connect_timeout_with_failover_ms, settings.send_timeout, settings.receive_timeout, settings.tcp_keep_alive_timeout);
     }
 
     static ConnectionTimeouts getHTTPTimeouts(const Settings & settings)

dbms/src/Interpreters/InterpreterSelectQuery.cpp

@@ -49,6 +49,7 @@
 #include <Parsers/queryToString.h>
 #include <ext/map.h>
 #include <memory>
+#include <DataStreams/ConvertingBlockInputStream.h>
 
 
 namespace DB
@@ -1293,6 +1294,17 @@ void InterpreterSelectQuery::executeUnion(Pipeline & pipeline)
     /// If there are still several streams, then we combine them into one
     if (pipeline.hasMoreThanOneStream())
     {
+        /// Unify streams in case they have different headers.
+        auto first_header = pipeline.streams.at(0)->getHeader();
+        for (size_t i = 1; i < pipeline.streams.size(); ++i)
+        {
+            auto & stream = pipeline.streams[i];
+            auto header = stream->getHeader();
+            auto mode = ConvertingBlockInputStream::MatchColumnsMode::Name;
+            if (!blocksHaveEqualStructure(first_header, header))
+                stream = std::make_shared<ConvertingBlockInputStream>(context, stream, first_header, mode);
+        }
+
         pipeline.firstStream() = std::make_shared<UnionBlockInputStream<>>(pipeline.streams, pipeline.stream_with_non_joined_data, max_streams);
         pipeline.stream_with_non_joined_data = nullptr;
         pipeline.streams.resize(1);

dbms/src/Interpreters/Settings.h

@@ -49,6 +49,7 @@ struct Settings
     M(SettingMilliseconds, connect_timeout_with_failover_ms, DBMS_DEFAULT_CONNECT_TIMEOUT_WITH_FAILOVER_MS, "Connection timeout for selecting first healthy replica.") \
     M(SettingSeconds, receive_timeout, DBMS_DEFAULT_RECEIVE_TIMEOUT_SEC, "") \
     M(SettingSeconds, send_timeout, DBMS_DEFAULT_SEND_TIMEOUT_SEC, "") \
+    M(SettingSeconds, tcp_keep_alive_timeout, 0, "") \
     M(SettingMilliseconds, queue_max_wait_ms, 5000, "The wait time in the request queue, if the number of concurrent requests exceeds the maximum.") \
     M(SettingUInt64, poll_interval, DBMS_DEFAULT_POLL_INTERVAL, "Block at the query wait loop on the server for the specified number of seconds.") \
     M(SettingUInt64, distributed_connections_pool_size, DBMS_DEFAULT_DISTRIBUTED_CONNECTIONS_POOL_SIZE, "Maximum number of connections with one remote server in the pool.") \

dbms/tests/integration/test_union_header/configs/remote_servers.xml

@@ -0,0 +1,18 @@
+<yandex>
+    <remote_servers>
+        <two_shards>
+            <shard>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+            <shard>
+                <replica>
+                    <host>node2</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </two_shards>
+    </remote_servers>
+</yandex>

dbms/tests/integration/test_union_header/test.py

@@ -0,0 +1,44 @@
+import pytest
+
+from helpers.cluster import ClickHouseCluster
+
+cluster = ClickHouseCluster(__file__)
+
+node1 = cluster.add_instance('node1', main_configs=['configs/remote_servers.xml'], with_zookeeper=True)
+node2 = cluster.add_instance('node2', main_configs=['configs/remote_servers.xml'], with_zookeeper=True)
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+
+        for node in (node1, node2):
+
+            node.query('''
+            CREATE TABLE default.t1_local
+            (
+                event_date Date DEFAULT toDate(event_time),
+                event_time DateTime,
+                log_type UInt32,
+                account_id String
+            )
+            ENGINE = MergeTree(event_date, (event_time, account_id), 8192);
+            ''')
+
+            node.query('''
+            CREATE TABLE default.t1 AS default.t1_local
+            ENGINE = Distributed('two_shards', 'default', 't1_local', rand());
+            ''')
+
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+
+def test_read(started_cluster):
+    assert node1.query('''SELECT event_date, event_time, log_type
+                        FROM default.t1
+                        WHERE (log_type = 30305) AND (account_id = '111111')
+                        LIMIT 1''').strip() == ''

dbms/tests/queries/0_stateless/00675_long_sql_fuzzy.reference

@@ -0,0 +1 @@
+Still alive

dbms/tests/queries/0_stateless/00746_sql_fuzzy.sh

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+. $CURDIR/../shell_config.sh
+
+export SQL_FUZZY_FILE_FUNCTIONS=${CLICKHOUSE_TMP}/clickhouse-functions
+$CLICKHOUSE_CLIENT -q "select name from system.functions format TSV;" > $SQL_FUZZY_FILE_FUNCTIONS
+
+export SQL_FUZZY_FILE_TABLE_FUNCTIONS=${CLICKHOUSE_TMP}/clickhouse-table_functions
+$CLICKHOUSE_CLIENT -q "select name from system.table_functions format TSV;" > $SQL_FUZZY_FILE_TABLE_FUNCTIONS
+
+# This is short run for ordinary tests.
+# if you want long run use: env SQL_FUZZY_RUNS=100000 clickhouse-test sql_fuzzy
+
+for SQL_FUZZY_RUN in $(seq ${SQL_FUZZY_RUNS:=100}); do
+    env SQL_FUZZY_RUN=$SQL_FUZZY_RUN $CURDIR/00746_sql_fuzzy.pl | $CLICKHOUSE_CLIENT -n --ignore-error >/dev/null 2>&1
+    if [[ `$CLICKHOUSE_CLIENT -q "SELECT 'Still alive'"` != 'Still alive' ]]; then
+        break
+    fi
+done
+
+$CLICKHOUSE_CLIENT -q "SELECT 'Still alive'"

dbms/tests/queries/bugs/fuzzy.sql

@@ -0,0 +1,5 @@
+SELECT __inner_restore_projection__(2.0885, -66.72488);
+SELECT __inner_restore_projection__(-4, '');
+SELECT __inner_restore_projection__(067274, 'vb\s');
+SELECT sequenceCount((CAST((( SELECT NULL ) AS rg, ( SELECT ( SELECT [], '<e', caseWithExpr([NULL], -588755.149, []), retention(addWeeks((CAST((-7644612.39732) AS DateTime)), -23578040.02833), (CAST(([]) AS DateTime)), (CAST(([010977.08]) AS String))), emptyArrayToSingle('') ) , '\0', toUInt64([], 't3hw@'), '\0', toStartOfQuarter(-4230.1872, []) ) ) AS Date)));
+SELECT extractURLParameter('?_', '\0');

docs/en/query_language/agg_functions/reference.md

@@ -53,6 +53,135 @@ FROM ontime
 Selects the last value encountered.
 The result is just as indeterminate as for the `any` function.
 
+##groupBitAnd
+
+Applies bitwise `AND` for series of numbers.
+
+```
+groupBitAnd(expr)
+```
+
+**Parameters**
+
+`expr` – An expression that results in `UInt*` type.
+
+**Return value**
+
+Value of the `UInt*` type.
+
+**Example**
+
+Test data:
+
+```
+binary     decimal
+00101100 = 44
+00011100 = 28
+00001101 = 13
+01010101 = 85
+```
+
+The query:
+
+```
+SELECT groupBitAnd(num) FROM t
+```
+
+Where `num` is the column with the test data.
+
+Result:
+
+```
+binary     decimal
+00000100 = 4
+```
+
+##groupBitOr
+
+Applies bitwise `OR` for series of numbers.
+
+```
+groupBitOr(expr)
+```
+
+**Parameters**
+
+`expr` – An expression that results in `UInt*` type.
+
+**Return value**
+
+Value of the `UInt*` type.
+
+**Example**
+
+Test data:
+
+```
+binary     decimal
+00101100 = 44
+00011100 = 28
+00001101 = 13
+01010101 = 85
+```
+
+Query:
+
+```
+SELECT groupBitOr(num) FROM t
+```
+
+Where `num` is the column with the test data.
+
+Result:
+
+```
+binary     decimal
+01111101 = 125
+```
+
+##groupBitXor
+
+Applies bitwise `XOR` for series of numbers.
+
+```
+groupBitXor(expr)
+```
+
+**Parameters**
+
+`expr` – An expression that results in `UInt*` type.
+
+**Return value**
+
+Value of the `UInt*` type.
+
+**Example**
+
+Test data:
+
+```
+binary     decimal
+00101100 = 44
+00011100 = 28
+00001101 = 13
+01010101 = 85
+```
+
+Query:
+
+```
+SELECT groupBitXor(num) FROM t
+```
+
+Where `num` is the column with the test data.
+
+Result:
+
+```
+binary     decimal
+01101000 = 104
+```
+
 ## min(x)
 
 Calculates the minimum.