dbms: prepared for fully-functional progress bar [#METR-2944].

dbms/include/DB/Common/formatReadable.h

@@ -0,0 +1,13 @@
+#pragma once
+
+#include <string>
+
+
+/// Выводит переданный размер в байтах в виде 123.45 GiB.
+std::string formatReadableSizeWithBinarySuffix(double value, int precision = 2);
+
+/// Выводит переданный размер в байтах в виде 132.55 GB.
+std::string formatReadableSizeWithDecimalSuffix(double value, int precision = 2);
+
+/// Выводит число в виде 123.45 billion.
+std::string formatReadableQuantity(double value, int precision = 2);

dbms/include/DB/Core/Defines.h

@@ -64,6 +64,7 @@
 #define DBMS_MIN_REVISION_WITH_TOTALS_EXTREMES					35265
 #define DBMS_MIN_REVISION_WITH_STRING_QUERY_ID					39002
 #define DBMS_MIN_REVISION_WITH_TEMPORARY_TABLES					50264
+#define DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS			51554
 
 #define DBMS_DISTRIBUTED_DIRECTORY_MONITOR_SLEEP_TIME_MS		100
 

dbms/include/DB/Core/Progress.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <DB/Core/Defines.h>
 #include <DB/IO/ReadBuffer.h>
 #include <DB/IO/WriteBuffer.h>
 #include <DB/IO/ReadHelpers.h>
@@ -10,25 +11,72 @@ namespace DB
 {
 
 
-/// Прогресс выполнения запроса
+/** Прогресс выполнения запроса.
+  * Передаваемые по сети значения представляют собой разницу - сколько было сделано после предыдущего отправленного значения.
+  * Тот же объект используется для суммирования полученных значений.
+  */
 struct Progress
 {
-	size_t rows;	/// Строк обработано.
-	size_t bytes;	/// Байт обработано.
+	size_t rows = 0;		/// Строк обработано.
+	size_t bytes = 0;		/// Байт обработано.
 
-	Progress() : rows(0), bytes(0) {}
-	Progress(size_t rows_, size_t bytes_) : rows(rows_), bytes(bytes_) {}
+	/** Сколько ещё строк надо обработать, приблизительно. Передаётся не ноль, когда возникает информация о какой-то новой части работы.
+	  * Полученные значения надо суммровать, чтобы получить оценку общего количества строк для обработки.
+	  * Используется для отображения прогресс-бара на клиенте.
+	  */
+	size_t total_rows = 0;
 
-	void read(ReadBuffer & in)
+	Progress() {}
+	Progress(size_t rows_, size_t bytes_, size_t total_rows_ = 0)
+		: rows(rows_), bytes(bytes_), total_rows(total_rows_) {}
+
+	void read(ReadBuffer & in, UInt64 server_revision)
 	{
 		readVarUInt(rows, in);
 		readVarUInt(bytes, in);
+
+		if (server_revision >= DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS)
+			readVarUInt(total_rows, in);
 	}
 
-	void write(WriteBuffer & out)
+	void write(WriteBuffer & out, UInt64 client_revision) const
 	{
 		writeVarUInt(rows, out);
 		writeVarUInt(bytes, out);
+
+		if (client_revision >= DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS)
+			writeVarUInt(total_rows, out);
+	}
+
+	void increment(const Progress & rhs)
+	{
+		rows += rhs.rows;
+		bytes += rhs.bytes;
+		total_rows += rhs.total_rows;
+	}
+
+	/// Каждое значение по-отдельности изменяется атомарно.
+	void incrementPiecewiseAtomically(const Progress & rhs)
+	{
+		__sync_add_and_fetch(&rows, rhs.rows);
+		__sync_add_and_fetch(&bytes, rhs.bytes);
+		__sync_add_and_fetch(&total_rows, rhs.total_rows);
+	}
+
+	void reset()
+	{
+		*this = Progress();
+	}
+
+	Progress fetchAndResetPiecewiseAtomically()
+	{
+		Progress res;
+
+		res.rows = __sync_fetch_and_and(&rows, 0);
+		res.bytes = __sync_fetch_and_and(&bytes, 0);
+		res.total_rows = __sync_fetch_and_and(&total_rows, 0);
+
+		return res;
 	}
 };
 

dbms/include/DB/DataStreams/IBlockInputStream.h

@@ -5,6 +5,7 @@
 #include <Poco/SharedPtr.h>
 
 #include <DB/Core/Block.h>
+#include <DB/Core/Progress.h>
 #include <DB/Storages/IStorage.h>
 
 
@@ -18,7 +19,7 @@ using Poco::SharedPtr;
   * Функция принимает количество строк в последнем блоке, количество байт в последнем блоке.
   * Следует иметь ввиду, что колбэк может вызываться из разных потоков.
   */
-typedef std::function<void(size_t, size_t)> ProgressCallback;
+typedef std::function<void(const Progress & progress)> ProgressCallback;
 
 
 /** Интерфейс потока для чтения данных по блокам из БД.
@@ -29,9 +30,9 @@ class IBlockInputStream : private boost::noncopyable
 public:
 	typedef SharedPtr<IBlockInputStream> BlockInputStreamPtr;
 	typedef std::vector<BlockInputStreamPtr> BlockInputStreams;
-	
+
 	IBlockInputStream() {}
-	
+
 	/** Прочитать следующий блок.
 	  * Если блоков больше нет - вернуть пустой блок (для которого operator bool возвращает false).
 	  */
@@ -61,7 +62,7 @@ public:
 	virtual String getID() const = 0;
 
 	BlockInputStreams & getChildren() { return children; }
-	
+
 	void dumpTree(std::ostream & ostr, size_t indent = 0, size_t multiplier = 1);
 
 	/// Получить листовые источники (не считая этот).

dbms/include/DB/DataStreams/IProfilingBlockInputStream.h

@@ -5,6 +5,7 @@
 #include <statdaemons/Stopwatch.h>
 
 #include <DB/Core/Names.h>
+#include <DB/Core/Progress.h>
 
 #include <DB/Interpreters/Limits.h>
 #include <DB/Interpreters/Quota.h>
@@ -109,8 +110,8 @@ public:
 	  * - проверяются ограничения и квоты, которые должны быть проверены не в рамках одного источника,
 	  *   а над общим количеством потраченных ресурсов во всех источниках сразу (информация в ProcessList-е).
 	  */
-	virtual void progress(size_t rows, size_t bytes) { progressImpl(rows, bytes); }
-	void progressImpl(size_t rows, size_t bytes);
+	virtual void progress(const Progress & value) { progressImpl(value); }
+	void progressImpl(const Progress & value);
 
 
 	/** Установить указатель на элемент списка процессов.

dbms/include/DB/DataStreams/RemoteBlockInputStream.h

@@ -66,7 +66,7 @@ public:
 	/** Отменяем умолчальное уведомление о прогрессе,
 	  * так как колбэк прогресса вызывается самостоятельно.
 	  */
-	void progress(size_t rows, size_t bytes) {}
+	void progress(const Progress & value) override {}
 
 
 	void cancel()
@@ -156,7 +156,7 @@ protected:
 					  *  ограничений (например, минимальная скорость выполнения запроса)
 					  *  и квот (например, на количество строчек для чтения).
 					  */
-					progressImpl(packet.progress.rows, packet.progress.bytes);
+					progressImpl(packet.progress);
 
 					if (!was_cancelled && !finished && isCancelled())
 						cancel();

dbms/include/DB/Interpreters/ProcessList.h

@@ -8,6 +8,7 @@
 #include <Poco/Net/IPAddress.h>
 #include <statdaemons/Stopwatch.h>
 #include <DB/Core/Defines.h>
+#include <DB/Core/Progress.h>
 #include <DB/Core/Exception.h>
 #include <DB/Core/ErrorCodes.h>
 #include <DB/Common/MemoryTracker.h>
@@ -35,8 +36,7 @@ public:
 
 		Stopwatch watch;
 
-		volatile size_t rows_processed = 0;
-		volatile size_t bytes_processed = 0;
+		Progress progress;
 
 		MemoryTracker memory_tracker;
 
@@ -56,10 +56,9 @@ public:
 			current_memory_tracker = nullptr;
 		}
 
-		bool update(size_t rows, size_t bytes) volatile
+		bool update(const Progress & value)
 		{
-			__sync_add_and_fetch(&rows_processed, rows);
-			__sync_add_and_fetch(&bytes_processed, bytes);
+			progress.incrementPiecewiseAtomically(value);
 			return !is_cancelled;
 		}
 	};

dbms/include/DB/Storages/MergeTree/MergeTreeBlockInputStream.h

@@ -68,8 +68,16 @@ public:
 			columns = owned_data_part->columns.addTypes(column_names);
 		}
 
+		/// Оценим общее количество строк - для прогресс-бара.
+		for (const auto & range : all_mark_ranges)
+			total_rows += range.end - range.begin;
+		total_rows *= storage.index_granularity;
+
 		LOG_TRACE(log, "Reading " << all_mark_ranges.size() << " ranges from part " << owned_data_part->name
-			<< ", up to " << (all_mark_ranges.back().end - all_mark_ranges.front().begin) * storage.index_granularity
+			<< ", approx. " << total_rows
+			<< (all_mark_ranges.size() > 1
+				? ", up to " + toString((all_mark_ranges.back().end - all_mark_ranges.front().begin) * storage.index_granularity)
+				: "")
 			<< " rows starting from " << all_mark_ranges.front().begin * storage.index_granularity);
 	}
 
@@ -97,7 +105,7 @@ public:
 
 protected:
 	/// Будем вызывать progressImpl самостоятельно.
-	void progress(size_t rows, size_t bytes) {}
+	void progress(const Progress & value) override {}
 
 	Block readImpl()
 	{
@@ -108,6 +116,10 @@ protected:
 
 		if (!reader)
 		{
+			/// Отправим информацию о том, что собираемся читать примерно столько-то строк.
+			/// NOTE В конструкторе это делать не получилось бы, потому что тогда ещё не установлен progress_callback.
+			progressImpl(Progress(0, 0, total_rows));
+
 			UncompressedCache * uncompressed_cache = use_uncompressed_cache ? storage.context.getUncompressedCache() : NULL;
 			reader.reset(new MergeTreeReader(path, owned_data_part->name, columns, uncompressed_cache, storage, all_mark_ranges));
 			if (prewhere_actions)
@@ -135,7 +147,7 @@ protected:
 					if (range.begin == range.end)
 						remaining_mark_ranges.pop_back();
 				}
-				progressImpl(res.rows(), res.bytes());
+				progressImpl(Progress(res.rows(), res.bytes()));
 				pre_reader->fillMissingColumns(res);
 
 				/// Вычислим выражение в PREWHERE.
@@ -164,7 +176,7 @@ protected:
 						reader->readRange(range.begin, range.end, res);
 					}
 
-					progressImpl(0, res.bytes() - pre_bytes);
+					progressImpl(Progress(0, res.bytes() - pre_bytes));
 				}
 				else if (ColumnUInt8 * column_vec = typeid_cast<ColumnUInt8 *>(&*column))
 				{
@@ -216,7 +228,7 @@ protected:
 						continue;
 					}
 
-					progressImpl(0, res.bytes() - pre_bytes);
+					progressImpl(Progress(0, res.bytes() - pre_bytes));
 
 					post_filter.resize(post_filter_pos);
 
@@ -259,7 +271,7 @@ protected:
 					remaining_mark_ranges.pop_back();
 			}
 
-			progressImpl(res.rows(), res.bytes());
+			progressImpl(Progress(res.rows(), res.bytes()));
 
 			reader->fillMissingColumns(res);
 		}
@@ -297,6 +309,7 @@ private:
 	ExpressionActionsPtr prewhere_actions;
 	String prewhere_column;
 	bool remove_prewhere_column;
+	size_t total_rows = 0;	/// Приблизительное общее количество строк - для прогресс-бара.
 
 	Logger * log;
 };

dbms/src/Client/Benchmark.cpp

@@ -239,9 +239,8 @@ private:
 		Stopwatch watch;
 		RemoteBlockInputStream stream(connection, query, nullptr);
 
-		size_t read_rows = 0;
-		size_t read_bytes = 0;
-		stream.setProgressCallback([&](size_t rows_inc, size_t bytes_inc) { read_rows += rows_inc; read_bytes += bytes_inc; });
+		Progress progress;
+		stream.setProgressCallback([&progress](const Progress & value) { progress.incrementPiecewiseAtomically(value); });
 
 		stream.readPrefix();
 		while (Block block = stream.read())
@@ -253,8 +252,8 @@ private:
 		double seconds = watch.elapsedSeconds();
 
 		Poco::ScopedLock<Poco::FastMutex> lock(mutex);
-		info_per_interval.add(seconds, read_rows, read_bytes, info.rows, info.bytes);
-		info_total.add(seconds, read_rows, read_bytes, info.rows, info.bytes);
+		info_per_interval.add(seconds, progress.rows, progress.bytes, info.rows, info.bytes);
+		info_total.add(seconds, progress.rows, progress.bytes, info.rows, info.bytes);
 	}
 
 

dbms/src/Client/Client.cpp

@@ -28,6 +28,8 @@
 #include <DB/Core/Types.h>
 #include <DB/Core/QueryProcessingStage.h>
 
+#include <DB/Common/formatReadable.h>
+
 #include <DB/IO/ReadBufferFromFileDescriptor.h>
 #include <DB/IO/WriteBufferFromFileDescriptor.h>
 #include <DB/IO/WriteBufferFromString.h>
@@ -121,8 +123,9 @@ private:
 
 	Stopwatch watch;
 
-	size_t rows_read_on_server = 0;
-	size_t bytes_read_on_server = 0;
+	/// С сервера периодически приходит информация, о том, сколько прочитано данных за прошедшее время.
+	Progress progress;
+
 	size_t written_progress_chars = 0;
 	bool written_first_block = false;
 
@@ -470,8 +473,7 @@ private:
 			return true;
 
 		processed_rows = 0;
-		rows_read_on_server = 0;
-		bytes_read_on_server = 0;
+		progress.reset();
 		written_progress_chars = 0;
 		written_first_block = false;
 
@@ -511,7 +513,7 @@ private:
 			std::cout << std::endl
 				<< processed_rows << " rows in set. Elapsed: " << watch.elapsedSeconds() << " sec. ";
 
-			if (rows_read_on_server >= 1000)
+			if (progress.rows >= 1000)
 				writeFinalProgress();
 
 			std::cout << std::endl << std::endl;
@@ -809,11 +811,9 @@ private:
 	}
 
 
-	void onProgress(const Progress & progress)
+	void onProgress(const Progress & value)
 	{
-		rows_read_on_server += progress.rows;
-		bytes_read_on_server += progress.bytes;
-
+		progress.increment(value);
 		writeProgress();
 	}
 
@@ -851,13 +851,20 @@ private:
 		std::stringstream message;
 		message << indicators[increment % 8]
 			<< std::fixed << std::setprecision(3)
-			<< " Progress: " << rows_read_on_server << " rows, " << bytes_read_on_server / 1000000.0 << " MB";
+			<< " Progress: ";
+
+		if (progress.total_rows)
+			message << (100.0 * progress.rows / progress.total_rows) << "%, ";
+
+		message
+			<< formatReadableQuantity(progress.rows) << " rows, "
+			<< formatReadableSizeWithDecimalSuffix(progress.bytes);
 
 		size_t elapsed_ns = watch.elapsed();
 		if (elapsed_ns)
 			message << " ("
-				<< rows_read_on_server * 1000000000.0 / elapsed_ns << " rows/s., "
-				<< bytes_read_on_server * 1000.0 / elapsed_ns << " MB/s.) ";
+				<< formatReadableQuantity(progress.rows * 1000000000.0 / elapsed_ns) << " rows/s., "
+				<< formatReadableSizeWithDecimalSuffix(progress.bytes * 1000000000.0 / elapsed_ns) << "/s.) ";
 		else
 			message << ". ";
 
@@ -869,13 +876,15 @@ private:
 
 	void writeFinalProgress()
 	{
-		std::cout << "Processed " << rows_read_on_server << " rows, " << bytes_read_on_server / 1000000.0 << " MB";
+		std::cout << "Processed "
+			<< formatReadableQuantity(progress.rows) << " rows, "
+			<< formatReadableSizeWithDecimalSuffix(progress.bytes);
 
 		size_t elapsed_ns = watch.elapsed();
 		if (elapsed_ns)
 			std::cout << " ("
-				<< rows_read_on_server * 1000000000.0 / elapsed_ns << " rows/s., "
-				<< bytes_read_on_server * 1000.0 / elapsed_ns << " MB/s.) ";
+				<< formatReadableQuantity(progress.rows * 1000000000.0 / elapsed_ns) << " rows/s., "
+				<< formatReadableSizeWithDecimalSuffix(progress.bytes * 1000000000.0 / elapsed_ns) << "/s.) ";
 		else
 			std::cout << ". ";
 	}

dbms/src/Client/Connection.cpp

@@ -470,7 +470,7 @@ Progress Connection::receiveProgress()
 	//LOG_TRACE(log_wrapper.get(), "Receiving progress (" << getServerAddress() << ")");
 
 	Progress progress;
-	progress.read(*in);
+	progress.read(*in, server_revision);
 	return progress;
 }
 

dbms/src/Common/MemoryTracker.cpp

@@ -1,30 +1,16 @@
 #include <Yandex/likely.h>
 #include <Yandex/logger_useful.h>
 #include <DB/Core/Exception.h>
+#include <DB/Common/formatReadable.h>
 #include <DB/IO/WriteHelpers.h>
 #include <iomanip>
 
 #include <DB/Common/MemoryTracker.h>
 
 
-static std::string formatReadableSize(double size)
-{
-	const char* units[] = {"B", "KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB"};
-	size_t i = 0;
-	while (i + 1 < sizeof(units) / sizeof(units[0]) &&
-		fabs(size) >= 1024)
-	{
-		size /= 1024;
-        ++i;
-    }
-    std::stringstream ss;
-    ss << std::fixed << std::setprecision(i) << size << ' ' << units[i];
-    return ss.str();
-}
-
 MemoryTracker::~MemoryTracker()
 {
-	LOG_DEBUG(&Logger::get("MemoryTracker"), "Peak memory usage for query: " << formatReadableSize(peak) << ".");
+	LOG_DEBUG(&Logger::get("MemoryTracker"), "Peak memory usage for query: " << formatReadableSizeWithBinarySuffix(peak) << ".");
 }
 
 void MemoryTracker::alloc(Int64 size)
@@ -34,9 +20,9 @@ void MemoryTracker::alloc(Int64 size)
 	if (unlikely(limit && will_be > limit))
 	{
 		free(size);
-		throw DB::Exception("Memory limit exceeded: would use " + formatReadableSize(will_be) + ""
+		throw DB::Exception("Memory limit exceeded: would use " + formatReadableSizeWithBinarySuffix(will_be) + ""
 			" (attempt to allocate chunk of " + DB::toString(size) + " bytes)"
-			", maximum: " + formatReadableSize(limit), DB::ErrorCodes::MEMORY_LIMIT_EXCEEDED);
+			", maximum: " + formatReadableSizeWithBinarySuffix(limit), DB::ErrorCodes::MEMORY_LIMIT_EXCEEDED);
 	}
 
 	if (will_be > peak)

dbms/src/Common/formatReadable.cpp

@@ -0,0 +1,36 @@
+#include <cmath>
+#include <sstream>
+#include <iomanip>
+
+#include <DB/Common/formatReadable.h>
+
+
+static std::string formatReadable(double size, int precision, const char ** units, size_t units_size, double delimiter)
+{
+	size_t i = 0;
+	for (; i + 1 < units_size && fabs(size) >= delimiter; ++i)
+		size /= delimiter;
+
+    std::stringstream ss;
+    ss << std::fixed << std::setprecision(precision) << size << units[i];
+    return ss.str();
+}
+
+
+std::string formatReadableSizeWithBinarySuffix(double value, int precision)
+{
+	const char * units[]  = {" B", " KiB", " MiB", " GiB", " TiB", " PiB", " EiB", " ZiB", " YiB"};
+	return formatReadable(value, precision, units, sizeof(units) / sizeof(units[0]), 1024);
+}
+
+std::string formatReadableSizeWithDecimalSuffix(double value, int precision)
+{
+	const char * units[] = {" B", " KB", " MB", " GB", " TB", " PB", " EB", " ZB", " YB"};
+	return formatReadable(value, precision, units, sizeof(units) / sizeof(units[0]), 1000);
+}
+
+std::string formatReadableQuantity(double value, int precision)
+{
+	const char * units[] = {"", " thousand", " million", " billion", " trillion", " quadrillion"};
+	return formatReadable(value, precision, units, sizeof(units) / sizeof(units[0]), 1000);
+}

dbms/src/DataStreams/IProfilingBlockInputStream.cpp

@@ -166,7 +166,7 @@ Block IProfilingBlockInputStream::read()
 		cancel();
 	}
 
-	progress(res.rowsInFirstColumn(), res.bytes());
+	progress(Progress(res.rowsInFirstColumn(), res.bytes()));
 
 	return res;
 }
@@ -295,36 +295,45 @@ void IProfilingBlockInputStream::checkQuota(Block & block)
 }
 
 
-void IProfilingBlockInputStream::progressImpl(size_t rows, size_t bytes)
+void IProfilingBlockInputStream::progressImpl(const Progress & value)
 {
 	/// Данные для прогресса берутся из листовых источников.
 	if (children.empty())
 	{
 		if (progress_callback)
-			progress_callback(rows, bytes);
+			progress_callback(value);
 
 		if (process_list_elem)
 		{
-			if (!process_list_elem->update(rows, bytes))
+			if (!process_list_elem->update(value))
 				cancel();
 
-			/// Общее количество данных, обработанных во всех листовых источниках, возможно, на удалённых серверах.
+			/// Общее количество данных, обработанных или предполагаемых к обработке во всех листовых источниках, возможно, на удалённых серверах.
 
-			size_t total_rows = process_list_elem->rows_processed;
-			size_t total_bytes = process_list_elem->bytes_processed;
+			size_t rows_processed = process_list_elem->progress.rows;
+			size_t bytes_processed = process_list_elem->progress.bytes;
+
+			size_t total_rows_estimate = std::max(process_list_elem->progress.rows, process_list_elem->progress.total_rows);
 
 			/** Проверяем ограничения на объём данных для чтения, скорость выполнения запроса, квоту на объём данных для чтения.
 			  * NOTE: Может быть, имеет смысл сделать, чтобы они проверялись прямо в ProcessList?
 			  */
 
 			if (limits.mode == LIMITS_TOTAL
-				&& ((limits.max_rows_to_read && total_rows > limits.max_rows_to_read)
-					|| (limits.max_bytes_to_read && total_bytes > limits.max_bytes_to_read)))
+				&& ((limits.max_rows_to_read && total_rows_estimate > limits.max_rows_to_read)
+					|| (limits.max_bytes_to_read && bytes_processed > limits.max_bytes_to_read)))
 			{
 				if (limits.read_overflow_mode == OverflowMode::THROW)
-					throw Exception("Limit for rows to read exceeded: read " + toString(total_rows)
-						+ " rows, maximum: " + toString(limits.max_rows_to_read),
-						ErrorCodes::TOO_MUCH_ROWS);
+				{
+					if (limits.max_rows_to_read && total_rows_estimate > limits.max_rows_to_read)
+						throw Exception("Limit for rows to read exceeded: " + toString(total_rows_estimate)
+							+ " rows read (or to read), maximum: " + toString(limits.max_rows_to_read),
+							ErrorCodes::TOO_MUCH_ROWS);
+					else
+						throw Exception("Limit for (uncompressed) bytes to read exceeded: " + toString(bytes_processed)
+							+ " bytes read, maximum: " + toString(limits.max_bytes_to_read),
+							ErrorCodes::TOO_MUCH_ROWS);
+				}
 				else if (limits.read_overflow_mode == OverflowMode::BREAK)
 					cancel();
 				else
@@ -336,9 +345,9 @@ void IProfilingBlockInputStream::progressImpl(size_t rows, size_t bytes)
 				double total_elapsed = info.total_stopwatch.elapsedSeconds();
 
 				if (total_elapsed > limits.timeout_before_checking_execution_speed.totalMicroseconds() / 1000000.0
-					&& total_rows / total_elapsed < limits.min_execution_speed)
+					&& rows_processed / total_elapsed < limits.min_execution_speed)
 				{
-					throw Exception("Query is executing too slow: " + toString(total_rows / total_elapsed)
+					throw Exception("Query is executing too slow: " + toString(rows_processed / total_elapsed)
 						+ " rows/sec., minimum: " + toString(limits.min_execution_speed),
 						ErrorCodes::TOO_SLOW);
 				}
@@ -346,7 +355,7 @@ void IProfilingBlockInputStream::progressImpl(size_t rows, size_t bytes)
 
 			if (quota != nullptr && limits.mode == LIMITS_TOTAL)
 			{
-				quota->checkAndAddReadRowsBytes(time(0), rows, bytes);
+				quota->checkAndAddReadRowsBytes(time(0), value.rows, value.bytes);
 			}
 		}
 	}

dbms/src/Server/TCPHandler.cpp

@@ -1,7 +1,5 @@
 #include <iomanip>
 
-#include <boost/bind.hpp>
-
 #include <Poco/Net/NetException.h>
 
 #include <Yandex/Revision.h>
@@ -85,9 +83,7 @@ void TCPHandler::runImpl()
 
 	sendHello();
 
-	connection_context.setProgressCallback([this] (const size_t rows, const size_t bytes) {
-		return this->updateProgress(rows, bytes);
-	});
+	connection_context.setProgressCallback([this] (const Progress & value) { return this->updateProgress(value); });
 
 	while (1)
 	{
@@ -125,6 +121,7 @@ void TCPHandler::runImpl()
 			/// Очищаем, так как, получая данные внешних таблиц, мы получили пустой блок.
 			/// А значит, stream помечен как cancelled и читать из него нельзя.
 			state.block_in = nullptr;
+			state.maybe_compressed_in = nullptr;	/// Для более корректного учёта MemoryTracker-ом.
 
 			/// Обрабатываем Query
 			state.io = executeQuery(state.query, query_context, false, state.stage);
@@ -286,7 +283,7 @@ void TCPHandler::processOrdinaryQuery()
 				}
 				else
 				{
-					if (state.rows_processed && after_send_progress.elapsed() / 1000 >= query_context.getSettingsRef().interactive_delay)
+					if (state.progress.rows && after_send_progress.elapsed() / 1000 >= query_context.getSettingsRef().interactive_delay)
 					{
 						/// Прошло некоторое время и есть прогресс.
 						after_send_progress.restart();
@@ -691,21 +688,17 @@ void TCPHandler::sendEndOfStream()
 }
 
 
-void TCPHandler::updateProgress(size_t rows, size_t bytes)
+void TCPHandler::updateProgress(const Progress & value)
 {
-	__sync_fetch_and_add(&state.rows_processed, rows);
-	__sync_fetch_and_add(&state.bytes_processed, bytes);
+	state.progress.incrementPiecewiseAtomically(value);
 }
 
 
 void TCPHandler::sendProgress()
 {
-	size_t rows_processed = __sync_fetch_and_and(&state.rows_processed, 0);
-	size_t bytes_processed = __sync_fetch_and_and(&state.bytes_processed, 0);
-
 	writeVarUInt(Protocol::Server::Progress, *out);
-	Progress progress(rows_processed, bytes_processed);
-	progress.write(*out);
+	Progress increment = state.progress.fetchAndResetPiecewiseAtomically();
+	increment.write(*out, client_revision);
 	out->next();
 }
 

dbms/src/Server/TCPHandler.h

@@ -23,8 +23,8 @@ struct QueryState
 	/// Идентификатор запроса.
 	String query_id;
 
-	QueryProcessingStage::Enum stage;
-	Protocol::Compression::Enum compression;
+	QueryProcessingStage::Enum stage = QueryProcessingStage::Complete;
+	Protocol::Compression::Enum compression = Protocol::Compression::Disable;
 
 	/// Откуда читать данные для INSERT-а.
 	SharedPtr<ReadBuffer> maybe_compressed_in;
@@ -40,24 +40,29 @@ struct QueryState
 	BlockIO io;
 
 	/// Отменен ли запрос
-	bool is_cancelled;
+	bool is_cancelled = false;
 	/// Пустой или нет
-	bool is_empty;
+	bool is_empty = true;
 	/// Данные были отправлены.
-	bool sent_all_data;
+	bool sent_all_data = false;
 	/// Запрос на вставку или нет.
-	bool is_insert;
+	bool is_insert = false;
 
 	/// Для вывода прогресса - разница после предыдущей отправки прогресса.
-	volatile size_t rows_processed;
-	volatile size_t bytes_processed;
+	Progress progress;
 
 
-	QueryState() : query_id(""), stage(QueryProcessingStage::Complete), compression(Protocol::Compression::Disable),
-		is_cancelled(false), is_empty(true), sent_all_data(false), is_insert(false), rows_processed(0), bytes_processed(0) {}
-	
 	void reset()
 	{
+		/** process_list_entry также включает/выключает учёт памяти MemoryTracker-ом.
+		  * Члены maybe_compressed_in, block_in, maybe_compressed_out, block_out
+		  *  могли быть инициализированы до io, и выделенная в них память могла не быть учтена MemoryTracker-ом.
+		  * Если эти члены будут уничтожены раньше, то освобождение памяти будет учтено MemoryTracker-ом,
+		  *  и вычисленный расход памяти может оказаться отрицательным (это не проблема, но некрасиво).
+		  * Поэтому, сначала уничтожим process_list_entry.
+		  */
+		io.process_list_entry = nullptr;
+
 		*this = QueryState();
 	}
 
@@ -133,7 +138,7 @@ private:
 	bool isQueryCancelled();
 
 	/// Эта функция вызывается из разных потоков.
-	void updateProgress(size_t rows, size_t bytes);
+	void updateProgress(const Progress & value);
 
 	/// Вывести информацию о скорости выполнения SELECT запроса.
 	void logProfileInfo(Stopwatch & watch, IBlockInputStream & in);

dbms/src/Storages/MergeTree/MergeTreeDataMerger.cpp

@@ -350,11 +350,13 @@ MergeTreeData::DataPartPtr MergeTreeDataMerger::mergeParts(
 		auto input = stdext::make_unique<MergeTreeBlockInputStream>(
 			data.getFullPath() + parts[i]->name + '/', DEFAULT_MERGE_BLOCK_SIZE, union_column_names, data,
 			parts[i], ranges, false, nullptr, "");
-		input->setProgressCallback([&merge_entry, rows_total] (const std::size_t rows, const std::size_t bytes) {
-			const auto new_rows_read = __sync_add_and_fetch(&merge_entry->rows_read, rows);
-			merge_entry->progress = static_cast<Float64>(new_rows_read) / rows_total;
-			__sync_add_and_fetch(&merge_entry->bytes_read_uncompressed, bytes);
-		});
+
+		input->setProgressCallback([&merge_entry, rows_total] (const Progress & value)
+			{
+				const auto new_rows_read = __sync_add_and_fetch(&merge_entry->rows_read, value.rows);
+				merge_entry->progress = static_cast<Float64>(new_rows_read) / rows_total;
+				__sync_add_and_fetch(&merge_entry->bytes_read_uncompressed, value.bytes);
+			});
 
 		src_streams.push_back(new ExpressionBlockInputStream(input.release(), data.getPrimaryExpression()));
 		sum_rows_approx += parts[i]->size * data.index_granularity;

dbms/src/Storages/StorageSystemProcesses.cpp

@@ -17,6 +17,7 @@ StorageSystemProcesses::StorageSystemProcesses(const std::string & name_, const
 		{ "elapsed", 		new DataTypeFloat64	},
 		{ "rows_read", 		new DataTypeUInt64	},
 		{ "bytes_read",		new DataTypeUInt64	},
+		{ "total_rows_approx", new DataTypeUInt64 },
 		{ "memory_usage",	new DataTypeUInt64	},
 		{ "query", 			new DataTypeString	},
 		{ "query_id", 		new DataTypeString	}
@@ -42,20 +43,19 @@ BlockInputStreams StorageSystemProcesses::read(
 	ColumnWithNameAndType col_elapsed{new ColumnFloat64, new DataTypeFloat64, "elapsed"};
 	ColumnWithNameAndType col_rows_read{new ColumnUInt64, new DataTypeUInt64, "rows_read"};
 	ColumnWithNameAndType col_bytes_read{new ColumnUInt64, new DataTypeUInt64, "bytes_read"};
+	ColumnWithNameAndType col_total_rows_approx{new ColumnUInt64, new DataTypeUInt64, "total_rows_approx"};
 	ColumnWithNameAndType col_memory_usage{new ColumnUInt64, new DataTypeUInt64, "memory_usage"};
 	ColumnWithNameAndType col_query{new ColumnString, new DataTypeString, "query"};
 	ColumnWithNameAndType col_query_id{new ColumnString, new DataTypeString, "query_id"};
 
 	for (const auto & process : context.getProcessList().get())
 	{
-		const size_t rows_read = process.rows_processed;
-		const size_t bytes_read = process.bytes_processed;
-
 		col_user.column->insert(process.user);
 		col_address.column->insert(process.ip_address.toString());
 		col_elapsed.column->insert(process.watch.elapsedSeconds());
-		col_rows_read.column->insert(rows_read);
-		col_bytes_read.column->insert(bytes_read);
+		col_rows_read.column->insert(process.progress.rows);
+		col_bytes_read.column->insert(process.progress.bytes);
+		col_total_rows_approx.column->insert(process.progress.total_rows);
 		col_memory_usage.column->insert(static_cast<UInt64>(process.memory_tracker.get()));
 		col_query.column->insert(process.query);
 		col_query_id.column->insert(process.query_id);
@@ -67,6 +67,7 @@ BlockInputStreams StorageSystemProcesses::read(
 		col_elapsed,
 		col_rows_read,
 		col_bytes_read,
+		col_total_rows_approx,
 		col_memory_usage,
 		col_query,
 		col_query_id