dbms: development [#CONV-2944].

dbms/include/DB/Core/Exception.h

@@ -19,6 +19,7 @@ public:
 	Exception(const std::string & msg, const std::string & arg, int code = 0);
 	Exception(const std::string & msg, const Exception & exc, int code = 0);
 	Exception(const Exception & exc);
+	explicit Exception(const Poco::Exception & exc);
 	~Exception() throw();
 	Exception & operator = (const Exception & exc);
 	const char * name() const throw();

dbms/include/DB/IO/ReadHelpers.h

@@ -432,4 +432,13 @@ inline void skipWhitespaceIfAny(ReadBuffer & buf)
 		++buf.position();
 }
 
+
+/** Прочитать сериализованный эксепшен.
+  * При сериализации/десериализации часть информации теряется
+  * (тип обрезается до базового, message заменяется на displayText, и stack trace дописывается в message)
+  * К нему может быть добавлено дополнительное сообщение (например, вы можете указать, откуда оно было прочитано).
+  */
+void readException(Exception & e, ReadBuffer & buf, const String & additional_message = "");
+void readAndThrowException(ReadBuffer & buf, const String & additional_message = "");
+
 }

dbms/include/DB/IO/WriteHelpers.h

@@ -437,4 +437,8 @@ inline void writeQuoted(const mysqlxx::DateTime & x,	WriteBuffer & buf)
 }
 
 
+/// Сериализация эксепшена (чтобы его можно было передать по сети)
+void writeException(const Exception & e, WriteBuffer & buf);
+
+
 }

dbms/src/Client/Client.cpp

@@ -1,3 +1,4 @@
+#include <unistd.h>
 #include <stdlib.h>
 #include <fcntl.h>
 
@@ -54,13 +55,16 @@ using Poco::SharedPtr;
 class Client : public Poco::Util::Application
 {
 public:
-	Client() : socket(), in(socket), out(socket), query_id(0), compression(Protocol::Compression::Enable),
+	Client() : is_interactive(true), stdin_is_not_tty(false), socket(), in(socket), out(socket), query_id(0), compression(Protocol::Compression::Enable),
 		format_max_block_size(0), std_out(STDOUT_FILENO), received_rows(0) {}
 	
 private:
 	typedef std::tr1::unordered_set<String> StringSet;
 	StringSet exit_strings;
 
+	bool is_interactive;				/// Использовать readline интерфейс или batch режим.
+	bool stdin_is_not_tty;				/// stdin - не терминал.
+
 	Poco::Net::StreamSocket socket;
 	ReadBufferFromPocoSocket in;
 	WriteBufferFromPocoSocket out;
@@ -115,7 +119,7 @@ private:
 
 		const char * home_path_cstr = getenv("HOME");
 		if (!home_path_cstr)
-			throw DB::Exception("Cannot get HOME environment variable");
+			throw Exception("Cannot get HOME environment variable");
 		else
 			home_path = home_path_cstr;
 
@@ -171,23 +175,36 @@ private:
 
 	int mainImpl(const std::vector<std::string> & args)
 	{
-		std::cout << std::fixed << std::setprecision(3);
+		/** Будем работать в batch режиме, если выполнено одно из следующих условий:
+		  * - задан параметр -e (--query)
+		  *   (в этом случае - запрос или несколько запросов берём оттуда;
+		  *    а если при этом stdin не терминал, то берём оттуда данные для INSERT-а первого запроса).
+		  * - stdin - не терминал (в этом случае, считываем оттуда запросы);
+		  */
+		stdin_is_not_tty = !isatty(STDIN_FILENO);
+		if (stdin_is_not_tty || config().has("query"))
+			is_interactive = false;
 		
-		std::cout << "ClickHouse client version " << DBMS_VERSION_MAJOR
+		std::cout << std::fixed << std::setprecision(3);
-			<< "." << DBMS_VERSION_MINOR
+		std::cerr << std::fixed << std::setprecision(3);
-			<< "." << Revision::get()
+		
-			<< "." << std::endl;
+		if (is_interactive)
+			std::cout << "ClickHouse client version " << DBMS_VERSION_MAJOR
+				<< "." << DBMS_VERSION_MINOR
+				<< "." << Revision::get()
+				<< "." << std::endl;
 
 		compression = config().getBool("compression", true) ? Protocol::Compression::Enable : Protocol::Compression::Disable;
 		in_format 	= config().getString("in_format", 	"Native");
 		out_format 	= config().getString("out_format", 	"Native");
-		format 		= config().getString("format", 		"Pretty");
+		format 		= config().getString("format", is_interactive ? "PrettyCompact" : "TabSeparated");
 		format_max_block_size = config().getInt("format_max_block_size", DEFAULT_BLOCK_SIZE);
 
 		String host = config().getString("host", "localhost");
 		UInt16 port = config().getInt("port", 9000);
-		
+
-		std::cout << "Connecting to " << host << ":" << port << "." << std::endl;
+		if (is_interactive)
+			std::cout << "Connecting to " << host << ":" << port << "." << std::endl;
 
 		socket.connect(Poco::Net::SocketAddress(host, port));
 
@@ -207,33 +224,39 @@ private:
 		readVarUInt(server_version_minor, in);
 		readVarUInt(server_revision, in);
 
-		std::cout << "Connected to " << server_name
+		if (is_interactive)
-			<< " server version " << server_version_major
+			std::cout << "Connected to " << server_name
-			<< "." << server_version_minor
+				<< " server version " << server_version_major
-			<< "." << server_revision
+				<< "." << server_version_minor
-			<< "." << std::endl << std::endl;
+				<< "." << server_revision
+				<< "." << std::endl << std::endl;
 
 		context.format_factory = new FormatFactory();
 		context.data_type_factory = new DataTypeFactory();
 
-		/// Отключаем tab completion.
+		if (is_interactive)
-		rl_bind_key('\t', rl_insert);
-
-		/// Загружаем историю команд, если есть.
-		history_file = config().getString("history_file", home_path + "/.clickhouse-client-history");
-		if (Poco::File(history_file).exists())
 		{
-			int res = read_history(history_file.c_str());
+			/// Отключаем tab completion.
-			if (res)
+			rl_bind_key('\t', rl_insert);
-				throwFromErrno("Cannot read history from file " + history_file, ErrorCodes::CANNOT_READ_HISTORY);
+
-		}
+			/// Загружаем историю команд, если есть.
-		else	/// Создаём файл с историей.
+			history_file = config().getString("history_file", home_path + "/.clickhouse-client-history");
-			Poco::File(history_file).createFile();
+			if (Poco::File(history_file).exists())
-		
+			{
-		loop();
+				int res = read_history(history_file.c_str());
+				if (res)
+					throwFromErrno("Cannot read history from file " + history_file, ErrorCodes::CANNOT_READ_HISTORY);
+			}
+			else	/// Создаём файл с историей.
+				Poco::File(history_file).createFile();
+
+			loop();
+
+			std::cout << "Bye." << std::endl;
+		}
+		else
+			nonInteractive();
 
-		std::cout << "Bye." << std::endl;
-		
 		return 0;
 	}
 
@@ -260,6 +283,17 @@ private:
 	}
 
 
+	void nonInteractive()
+	{
+		if (config().has("query"))
+			process(config().getString("query"));
+		else
+		{
+			
+		}
+	}
+
+
 	bool process(const String & line)
 	{
 		if (exit_strings.end() != exit_strings.find(line))
@@ -276,9 +310,10 @@ private:
 		sendQuery();
 		receiveResult();
 
-		std::cout << std::endl
+		if (is_interactive)
-			<< received_rows << " rows in set. Elapsed: " << watch.elapsedSeconds() << " sec."
+			std::cout << std::endl
-			<< std::endl << std::endl;
+				<< received_rows << " rows in set. Elapsed: " << watch.elapsedSeconds() << " sec."
+				<< std::endl << std::endl;
 
 		block_in = NULL;
 		maybe_compressed_in = NULL;
@@ -302,7 +337,7 @@ private:
 		/// Распарсенный запрос должен заканчиваться на конец входных данных или на точку с запятой.
 		if (!parse_res || (pos != end && *pos != ';'))
 		{
-			std::cout << "Syntax error: failed at position "
+			std::cerr << "Syntax error: failed at position "
 				<< (pos - begin) << ": "
 				<< std::string(pos, std::min(SHOW_CHARS_ON_SYNTAX_ERROR, end - pos))
 				<< ", expected " << (parse_res ? "end of query" : expected) << "."
@@ -311,9 +346,12 @@ private:
 			return false;
 		}
 
-		std::cout << std::endl;
+		if (is_interactive)
-		formatAST(*parsed_query, std::cout);
+		{
-		std::cout << std::endl << std::endl;
+			std::cout << std::endl;
+			formatAST(*parsed_query, std::cout);
+			std::cout << std::endl << std::endl;
+		}
 		
 		return true;
 	}
@@ -356,6 +394,10 @@ private:
 			case Protocol::Server::Data:
 				return receiveData();
 
+			case Protocol::Server::Exception:
+				receiveException();
+				return false;
+
 			default:
 				throw Exception("Unknown packet from server", ErrorCodes::UNKNOWN_PACKET_FROM_SERVER);
 		}
@@ -404,6 +446,16 @@ private:
 		else
 			return false;
 	}
+
+
+	void receiveException()
+	{
+		Exception e;
+		readException(e, in);
+
+		std::cerr << "Received exception from server:" << std::endl
+			<< "Code: " << e.code() << ". " << e.displayText();
+	}
 	
 
 	void defineOptions(Poco::Util::OptionSet & options)
@@ -430,6 +482,13 @@ private:
 				.repeatable(false)
 				.argument("<number>")
 				.binding("port"));
+
+		options.addOption(
+			Poco::Util::Option("query", "e")
+				.required(false)
+				.repeatable(false)
+				.argument("<string>")
+				.binding("query"));
 	}
 };
 

dbms/src/Core/Exception.cpp

@@ -9,6 +9,7 @@ Exception::Exception(const std::string & msg, int code) : Poco::Exception(msg, c
 Exception::Exception(const std::string & msg, const std::string & arg, int code): Poco::Exception(msg, arg, code) {}
 Exception::Exception(const std::string & msg, const Exception & exc, int code): Poco::Exception(msg, exc, code), trace(exc.trace) {}
 Exception::Exception(const Exception & exc) : Poco::Exception(exc), trace(exc.trace) {}
+Exception::Exception(const Poco::Exception & exc) : Poco::Exception(exc) {}
 Exception::~Exception() throw() {}
 
 Exception & Exception::operator=(const Exception& exc)

dbms/src/IO/ReadHelpers.cpp

@@ -1,5 +1,8 @@
+#include <sstream>
+
 #include <DB/IO/ReadHelpers.h>
 
+
 namespace DB
 {
 
@@ -121,4 +124,47 @@ void readBackQuotedString(String & s, ReadBuffer & buf)
 	readAnyQuotedString<'`'>(s, buf);
 }
 
+
+void readException(Exception & e, ReadBuffer & buf, const String & additional_message)
+{
+	int code = 0;
+	String name;
+	String message;
+	String stack_trace;
+	bool has_nested = false;
+	
+	readBinary(code, buf);
+	readBinary(name, buf);
+	readBinary(message, buf);
+	readBinary(stack_trace, buf);
+	readBinary(has_nested, buf);
+
+	std::stringstream message_stream;
+
+	if (!additional_message.empty())
+		message_stream << additional_message << ". ";
+
+	if (name != "DB::Exception")
+		message_stream << name << ". ";
+
+	message_stream << message
+		<< ". Stack trace:\n\n" << stack_trace;
+
+	if (has_nested)
+	{
+		Exception nested;
+		readException(nested, buf);
+		e = Exception(message_stream.str(), nested, code);
+	}
+	else
+		e = Exception(message_stream.str(), code);
+}
+
+void readAndThrowException(ReadBuffer & buf, const String & additional_message)
+{
+	Exception e;
+	readException(e, buf, additional_message);
+	e.rethrow();
+}
+
 }

dbms/src/IO/WriteHelpers.cpp

@@ -3,4 +3,18 @@
 namespace DB
 {
 
+void writeException(const Exception & e, WriteBuffer & buf)
+{
+	writeBinary(e.code(), buf);
+	writeBinary(String(e.name()), buf);
+	writeBinary(e.message(), buf);
+	writeBinary(e.getStackTrace().toString(), buf);
+
+	bool has_nested = e.nested() != NULL;
+	writeBinary(has_nested, buf);
+
+	if (has_nested)
+		writeException(Exception(*e.nested()), buf);
+}
+
 }

dbms/src/Server/TCPHandler.cpp

@@ -32,29 +32,58 @@ void TCPHandler::runImpl()
 
 	while (!in.eof())
 	{
-		/// Пакет с запросом.
-		receivePacket(in);
-
-		LOG_DEBUG(log, "Query ID: " << state.query_id);
-		LOG_DEBUG(log, "Query: " << state.query);
-		LOG_DEBUG(log, "In format: " << state.in_format);
-		LOG_DEBUG(log, "Out format: " << state.out_format);
-
 		Stopwatch watch;
-
+		
-		/// Читаем из сети данные для INSERT-а, если надо, и вставляем их.
+		try
-		if (state.io.out)
 		{
-			while (receivePacket(in))
+			/// Пакет с запросом.
-				;
+			receivePacket(in);
+
+			LOG_DEBUG(log, "Query ID: " << state.query_id);
+			LOG_DEBUG(log, "Query: " << state.query);
+			LOG_DEBUG(log, "In format: " << state.in_format);
+			LOG_DEBUG(log, "Out format: " << state.out_format);
+
+			state.exception = NULL;
+		
+			/// Читаем из сети данные для INSERT-а, если надо, и вставляем их.
+			if (state.io.out)
+			{
+				while (receivePacket(in))
+					;
+			}
+
+			/// Вынимаем результат выполнения запроса, если есть, и пишем его в сеть.
+			if (state.io.in)
+			{
+				while (sendData(out, out_for_chunks))
+					;
+			}
+		}
+		catch (DB::Exception & e)
+		{
+			LOG_ERROR(log, "DB::Exception. Code: " << e.code() << ", e.displayText() = " << e.displayText()
+				<< ", Stack trace:\n\n" << e.getStackTrace().toString());
+			state.exception = e.clone();
+		}
+		catch (Poco::Exception & e)
+		{
+			LOG_ERROR(log, "Poco::Exception. Code: " << ErrorCodes::POCO_EXCEPTION << ", e.code() = " << e.code() << ", e.displayText() = " << e.displayText());
+			state.exception = new Exception(e.message(), e.code());
+		}
+		catch (std::exception & e)
+		{
+			LOG_ERROR(log, "std::exception. Code: " << ErrorCodes::STD_EXCEPTION << ", e.what() = " << e.what());
+			state.exception = new Exception(e.what(), ErrorCodes::STD_EXCEPTION);
+		}
+		catch (...)
+		{
+			LOG_ERROR(log, "Unknown exception. Code: " << ErrorCodes::UNKNOWN_EXCEPTION);
+			state.exception = new Exception("Unknown exception", ErrorCodes::UNKNOWN_EXCEPTION);
 		}
 
-		/// Вынимаем результат выполнения запроса, если есть, и пишем его в сеть.
+		if (state.exception)
-		if (state.io.in)
+			sendException(out);
-		{
-			while (sendData(out, out_for_chunks))
-				;
-		}
 
 		watch.stop();
 
@@ -187,7 +216,9 @@ bool TCPHandler::sendData(WriteBuffer & out, WriteBuffer & out_for_chunks)
 
 void TCPHandler::sendException(WriteBuffer & out)
 {
-	/// TODO
+	writeVarUInt(Protocol::Server::Exception, out);
+	writeException(*state.exception, out);
+	out.next();
 }
 
 void TCPHandler::sendProgress(WriteBuffer & out)

dbms/src/Server/TCPHandler.h

@@ -44,6 +44,12 @@ struct QueryState
 
 	Context context;
 
+	/** Исключение во время выполнения запроса (его надо отдать по сети клиенту).
+	  * Клиент сможет его принять, если оно не произошло во время отправки другого пакета.
+	  */
+	SharedPtr<Exception> exception;
+	
+
 	QueryState() : query_id(0), stage(Protocol::QueryProcessingStage::Complete), compression(Protocol::Compression::Disable) {}
 	
 	void reset()