Basic code quality of LiveView

dbms/CMakeLists.txt

@@ -114,6 +114,7 @@ add_headers_and_sources(dbms src/Columns)
 add_headers_and_sources(dbms src/Storages)
 add_headers_and_sources(dbms src/Storages/Distributed)
 add_headers_and_sources(dbms src/Storages/MergeTree)
+add_headers_and_sources(dbms src/Storages/LiveView)
 add_headers_and_sources(dbms src/Client)
 add_headers_and_sources(dbms src/Formats)
 add_headers_and_sources(dbms src/Processors)

dbms/src/Core/Settings.h

@@ -342,6 +342,8 @@ struct Settings : public SettingsCollection<Settings>
     M(SettingUInt64, max_partitions_per_insert_block, 100, "Limit maximum number of partitions in single INSERTed block. Zero means unlimited. Throw exception if the block contains too many partitions. This setting is a safety threshold, because using large number of partitions is a common misconception.") \
     M(SettingBool, check_query_single_value_result, true, "Return check query result as single 1/0 value") \
     \
+    M(SettingBool, allow_experimental_live_view, false, "Enable LIVE VIEW. Not mature enough.") \
+    \
     /** Obsolete settings that do nothing but left for compatibility reasons. Remove each one after half a year of obsolescence. */ \
     \
     M(SettingBool, allow_experimental_low_cardinality_type, true, "Obsolete setting, does nothing. Will be removed after 2019-08-13") \

dbms/src/DataStreams/PushingToViewsBlockOutputStream.cpp

@@ -12,7 +12,7 @@
 #include <Common/ThreadPool.h>
 #include <Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.h>
 #include <Storages/StorageValues.h>
-#include <Storages/StorageLiveView.h>
+#include <Storages/LiveView/StorageLiveView.h>
 
 namespace DB
 {
@@ -106,8 +106,7 @@ void PushingToViewsBlockOutputStream::write(const Block & block)
 
     if (auto * live_view = dynamic_cast<StorageLiveView *>(storage.get()))
     {
-        BlockOutputStreamPtr output_ = std::make_shared<LiveViewBlockOutputStream>(*live_view);
-        StorageLiveView::writeIntoLiveView(*live_view, block, context, output_);
+        StorageLiveView::writeIntoLiveView(*live_view, block, context);
     }
     else
     {

dbms/src/DataStreams/PushingToViewsBlockOutputStream.h

@@ -5,7 +5,6 @@
 #include <DataStreams/OneBlockInputStream.h>
 #include <DataStreams/MaterializingBlockInputStream.h>
 #include <Storages/StorageMaterializedView.h>
-#include <Storages/StorageLiveView.h>
 
 namespace DB
 {

dbms/src/Interpreters/InterpreterAlterQuery.cpp

@@ -8,9 +8,9 @@
 #include <Storages/AlterCommands.h>
 #include <Storages/MutationCommands.h>
 #include <Storages/PartitionCommands.h>
-#include <Storages/LiveViewCommands.h>
+#include <Storages/LiveView/LiveViewCommands.h>
+#include <Storages/LiveView/StorageLiveView.h>
 #include <Common/typeid_cast.h>
-#include <Storages/StorageLiveView.h>
 
 #include <algorithm>
 

dbms/src/Interpreters/InterpreterWatchQuery.cpp

@@ -25,6 +25,7 @@ namespace ErrorCodes
     extern const int UNKNOWN_STORAGE;
     extern const int UNKNOWN_TABLE;
     extern const int TOO_MANY_COLUMNS;
+    extern const int SUPPORT_IS_DISABLED;
 }
 
 BlockInputStreamPtr InterpreterWatchQuery::executeImpl()
@@ -34,6 +35,9 @@ BlockInputStreamPtr InterpreterWatchQuery::executeImpl()
 
 BlockIO InterpreterWatchQuery::execute()
 {
+    if (!context.getSettingsRef().allow_experimental_live_view)
+        throw Exception("Experimental LIVE VIEW feature is not enabled (the setting 'allow_experimental_live_view')", ErrorCodes::SUPPORT_IS_DISABLED);
+
     BlockIO res;
     const ASTWatchQuery & query = typeid_cast<const ASTWatchQuery &>(*query_ptr);
     String database;

dbms/src/DataStreams/LiveViewBlockInputStream.h → dbms/src/Storages/LiveView/LiveViewBlockInputStream.h

@@ -1,23 +1,6 @@
-/* Copyright (c) 2018 BlackBerry Limited
-
-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. */
 #pragma once
 
-#include <limits>
-
-#include <Common/ConcurrentBoundedQueue.h>
-#include <Poco/Condition.h>
-#include <DataStreams/OneBlockInputStream.h>
 #include <DataStreams/IBlockInputStream.h>
-#include <Storages/StorageLiveView.h>
 
 
 namespace DB
@@ -61,8 +44,8 @@ using NonBlockingResult = std::pair<Block, bool>;
         if (isCancelled() || storage->is_dropped)
             return;
         IBlockInputStream::cancel(kill);
-        Poco::FastMutex::ScopedLock lock(storage->mutex);
-        storage->condition.broadcast();
+        std::lock_guard lock(storage->mutex);
+        storage->condition.notify_all();
     }
 
     Block getHeader() const override { return storage->getHeader(); }
@@ -92,22 +75,22 @@ using NonBlockingResult = std::pair<Block, bool>;
 
     NonBlockingResult tryRead()
     {
-        return tryRead_(false);
+        return tryReadImpl(false);
     }
 
 protected:
     Block readImpl() override
     {
         /// try reading
-        return tryRead_(true).first;
+        return tryReadImpl(true).first;
     }
 
     /** tryRead method attempts to read a block in either blocking
      *  or non-blocking mode. If blocking is set to false
      *  then method return empty block with flag set to false
      *  to indicate that method would block to get the next block.
      */
-    NonBlockingResult tryRead_(bool blocking)
+    NonBlockingResult tryReadImpl(bool blocking)
     {
         Block res;
 
@@ -118,7 +101,7 @@ using NonBlockingResult = std::pair<Block, bool>;
         /// If blocks were never assigned get blocks
         if (!blocks)
         {
-            Poco::FastMutex::ScopedLock lock(storage->mutex);
+            std::lock_guard lock(storage->mutex);
             if (!active)
                 return { Block(), false };
             blocks = (*blocks_ptr);
@@ -135,7 +118,7 @@ using NonBlockingResult = std::pair<Block, bool>;
         if (it == end)
         {
             {
-                Poco::FastMutex::ScopedLock lock(storage->mutex);
+                std::unique_lock lock(storage->mutex);
                 if (!active)
                     return { Block(), false };
                 /// If we are done iterating over our blocks
@@ -162,7 +145,10 @@ using NonBlockingResult = std::pair<Block, bool>;
                     while (true)
                     {
                         UInt64 timestamp_usec = static_cast<UInt64>(timestamp.epochMicroseconds());
-                        bool signaled = storage->condition.tryWait(storage->mutex, std::max(static_cast<UInt64>(0), heartbeat_interval_usec - (timestamp_usec - last_event_timestamp_usec)) / 1000);
+
+                        /// Or spurious wakeup.
+                        bool signaled = std::cv_status::no_timeout == storage->condition.wait_for(lock,
+                            std::chrono::microseconds(std::max(UInt64(0), heartbeat_interval_usec - (timestamp_usec - last_event_timestamp_usec))));
 
                         if (isCancelled() || storage->is_dropped)
                         {
@@ -181,7 +167,7 @@ using NonBlockingResult = std::pair<Block, bool>;
                     }
                 }
             }
-            return tryRead_(blocking);
+            return tryReadImpl(blocking);
         }
 
         res = *it;

dbms/src/Storages/LiveView/LiveViewBlockOutputStream.h

@@ -0,0 +1,74 @@
+#pragma once
+
+#include <DataTypes/DataTypesNumber.h>
+#include <DataStreams/IBlockOutputStream.h>
+#include <Storages/LiveView/StorageLiveView.h>
+
+
+namespace DB
+{
+
+class LiveViewBlockOutputStream : public IBlockOutputStream
+{
+public:
+    explicit LiveViewBlockOutputStream(StorageLiveView & storage_) : storage(storage_) {}
+
+    void writePrefix() override
+    {
+        new_blocks = std::make_shared<Blocks>();
+        new_blocks_metadata = std::make_shared<BlocksMetadata>();
+        new_hash = std::make_shared<SipHash>();
+    }
+
+    void writeSuffix() override
+    {
+        UInt128 key;
+        String key_str;
+
+        new_hash->get128(key.low, key.high);
+        key_str = key.toHexString();
+
+        std::lock_guard lock(storage.mutex);
+
+        if (storage.getBlocksHashKey() != key_str)
+        {
+            new_blocks_metadata->hash = key_str;
+            new_blocks_metadata->version = storage.getBlocksVersion() + 1;
+
+            for (auto & block : *new_blocks)
+            {
+                block.insert({DataTypeUInt64().createColumnConst(
+                    block.rows(), new_blocks_metadata->version)->convertToFullColumnIfConst(),
+                    std::make_shared<DataTypeUInt64>(),
+                    "_version"});
+            }
+
+            (*storage.blocks_ptr) = new_blocks;
+            (*storage.blocks_metadata_ptr) = new_blocks_metadata;
+
+            storage.condition.notify_all();
+        }
+
+        new_blocks.reset();
+        new_blocks_metadata.reset();
+        new_hash.reset();
+    }
+
+    void write(const Block & block) override
+    {
+        new_blocks->push_back(block);
+        block.updateHash(*new_hash);
+    }
+
+    Block getHeader() const override { return storage.getHeader(); }
+
+private:
+    using SipHashPtr = std::shared_ptr<SipHash>;
+
+    BlocksPtr new_blocks;
+    BlocksMetadataPtr new_blocks_metadata;
+    SipHashPtr new_hash;
+    StorageLiveView & storage;
+};
+
+}

dbms/src/Storages/LiveViewCommands.h → dbms/src/Storages/LiveView/LiveViewCommands.h

@@ -12,9 +12,9 @@ limitations under the License. */
 
 #pragma once
 
-#include <Parsers/ASTAlterQuery.h>
 #include <optional>
-#include <Storages/StorageLiveView.h>
+#include <Parsers/ASTAlterQuery.h>
+#include <Storages/LiveView/StorageLiveView.h>
 
 namespace DB
 {

dbms/src/DataStreams/LiveViewEventsBlockInputStream.h → dbms/src/Storages/LiveView/LiveViewEventsBlockInputStream.h

@@ -9,19 +9,17 @@ 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. */
-#pragma once
 
-#include <limits>
+#pragma once
 
-#include <Common/ConcurrentBoundedQueue.h>
 #include <Poco/Condition.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <DataTypes/DataTypeString.h>
 #include <Columns/ColumnString.h>
 #include <Columns/ColumnsNumber.h>
 #include <DataStreams/OneBlockInputStream.h>
 #include <DataStreams/IBlockInputStream.h>
-#include <Storages/StorageLiveView.h>
+#include <Storages/LiveView/StorageLiveView.h>
 
 
 namespace DB
@@ -66,8 +64,8 @@ using NonBlockingResult = std::pair<Block, bool>;
         if (isCancelled() || storage->is_dropped)
             return;
         IBlockInputStream::cancel(kill);
-        Poco::FastMutex::ScopedLock lock(storage->mutex);
-        storage->condition.broadcast();
+        std::lock_guard lock(storage->mutex);
+        storage->condition.notify_all();
     }
 
     Block getHeader() const override
@@ -103,7 +101,7 @@ using NonBlockingResult = std::pair<Block, bool>;
 
     NonBlockingResult tryRead()
     {
-        return tryRead_(false);
+        return tryReadImpl(false);
     }
 
     Block getEventBlock()
@@ -120,15 +118,15 @@ using NonBlockingResult = std::pair<Block, bool>;
     Block readImpl() override
     {
         /// try reading
-        return tryRead_(true).first;
+        return tryReadImpl(true).first;
     }
 
     /** tryRead method attempts to read a block in either blocking
      *  or non-blocking mode. If blocking is set to false
      *  then method return empty block with flag set to false
      *  to indicate that method would block to get the next block.
      */
-    NonBlockingResult tryRead_(bool blocking)
+    NonBlockingResult tryReadImpl(bool blocking)
     {
         if (has_limit && num_updates == static_cast<Int64>(limit))
         {
@@ -137,7 +135,7 @@ using NonBlockingResult = std::pair<Block, bool>;
         /// If blocks were never assigned get blocks
         if (!blocks)
         {
-            Poco::FastMutex::ScopedLock lock(storage->mutex);
+            std::lock_guard lock(storage->mutex);
             if (!active)
                 return { Block(), false };
             blocks = (*blocks_ptr);
@@ -155,7 +153,7 @@ using NonBlockingResult = std::pair<Block, bool>;
         if (it == end)
         {
             {
-                Poco::FastMutex::ScopedLock lock(storage->mutex);
+                std::unique_lock lock(storage->mutex);
                 if (!active)
                     return { Block(), false };
                 /// If we are done iterating over our blocks
@@ -183,7 +181,10 @@ using NonBlockingResult = std::pair<Block, bool>;
                     while (true)
                     {
                         UInt64 timestamp_usec = static_cast<UInt64>(timestamp.epochMicroseconds());
-                        bool signaled = storage->condition.tryWait(storage->mutex, std::max(static_cast<UInt64>(0), heartbeat_interval_usec - (timestamp_usec - last_event_timestamp_usec)) / 1000);
+
+                        /// Or spurious wakeup.
+                        bool signaled = std::cv_status::no_timeout == storage->condition.wait_for(lock,
+                            std::chrono::microseconds(std::max(UInt64(0), heartbeat_interval_usec - (timestamp_usec - last_event_timestamp_usec))));
 
                         if (isCancelled() || storage->is_dropped)
                         {
@@ -202,7 +203,7 @@ using NonBlockingResult = std::pair<Block, bool>;
                     }
                 }
             }
-            return tryRead_(blocking);
+            return tryReadImpl(blocking);
         }
 
         // move right to the end