diff --git a/CMakeLists.txt b/CMakeLists.txt
index 1cee6e26..336d21c4 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -74,7 +74,6 @@ add_executable(
 	test/capabilities.cpp
 	test/auth.cpp
 	test/metadata.cpp
-	test/datetime.cpp
 )
 target_link_libraries(
 	unittests
diff --git a/include/mysql/connection.hpp b/include/mysql/connection.hpp
index c2d0cc04..7d2e3695 100644
--- a/include/mysql/connection.hpp
+++ b/include/mysql/connection.hpp
@@ -5,6 +5,7 @@
 #include "mysql/impl/handshake.hpp"
 #include "mysql/impl/basic_types.hpp"
 #include "mysql/error.hpp"
+#include "mysql/resultset.hpp"
 
 namespace mysql
 {
@@ -14,8 +15,10 @@ using connection_params = detail::handshake_params; // TODO: do we think this in
 template <typename Stream, typename Allocator=std::allocator<std::uint8_t>>
 class connection
 {
+	using channel_type = detail::channel<Stream>;
+
 	Stream next_level_;
-	detail::channel<Stream> channel_;
+	channel_type channel_;
 	detail::bytestring<Allocator> buffer_;
 public:
 	template <typename... Args>
@@ -35,7 +38,12 @@ public:
 	BOOST_ASIO_INITFN_RESULT_TYPE(CompletionToken, void(error_code))
 	async_handshake(const connection_params& params, CompletionToken&& token);
 
+	resultset<channel_type, Allocator> query(std::string_view query_string, error_code&);
+	resultset<channel_type, Allocator> query(std::string_view query_string);
 
+	template <typename CompletionToken>
+	BOOST_ASIO_INITFN_RESULT_TYPE(CompletionToken, void(error_code, resultset<channel_type, Allocator>))
+	async_query(std::string_view query_string, CompletionToken&& token);
 };
 
 }
diff --git a/include/mysql/datetime.hpp b/include/mysql/datetime.hpp
deleted file mode 100644
index ec5257ba..00000000
--- a/include/mysql/datetime.hpp
+++ /dev/null
@@ -1,59 +0,0 @@
-#ifndef INCLUDE_MYSQL_DATETIME_HPP_
-#define INCLUDE_MYSQL_DATETIME_HPP_
-
-#include <mysql/datetime.hpp>
-#include <cstdint>
-#include <string_view>
-#include <ostream>
-
-namespace mysql
-{
-
-class datetime
-{
-	std::uint16_t year_ {};
-	std::uint8_t month_ {};
-	std::uint8_t day_ {};
-	std::uint8_t hour_ {};
-	std::uint8_t minute_ {};
-	std::uint8_t second_ {};
-	std::uint32_t microsecond_ {};
-public:
-	constexpr datetime() = default;
-	constexpr datetime(std::uint16_t years, std::uint8_t month, std::uint8_t day,
-			 std::uint8_t hour=0, std::uint8_t min=0, std::uint8_t seconds=0, std::uint32_t micro=0) noexcept;
-
-	constexpr std::uint16_t year() const noexcept { return year_; }
-	constexpr std::uint8_t month() const noexcept { return month_; }
-	constexpr std::uint8_t day() const noexcept { return day_; }
-	constexpr std::uint8_t hour() const noexcept { return hour_; }
-	constexpr std::uint8_t minute() const noexcept { return minute_; }
-	constexpr std::uint8_t second() const noexcept { return second_; }
-	constexpr std::uint32_t microsecond() const noexcept { return microsecond_; }
-
-	constexpr void set_year(std::uint16_t value) noexcept { year_ = value; }
-	constexpr void set_month(std::uint8_t value) noexcept { month_ = value; }
-	constexpr void set_day(std::uint8_t value) noexcept { day_ = value; }
-	constexpr void set_hour(std::uint8_t value) noexcept { hour_ = value; }
-	constexpr void set_minute(std::uint8_t value) noexcept { minute_ = value; }
-	constexpr void set_second(std::uint8_t value) noexcept { second_ = value; }
-	constexpr void set_microsecond(std::uint32_t value) noexcept { microsecond_ = value; }
-
-	constexpr bool operator==(const datetime& rhs) const noexcept;
-	constexpr bool operator!=(const datetime& rhs) const noexcept { return !(*this==rhs); }
-
-	static constexpr std::size_t max_string_size = 4 + 2*5 + 6 + 6 + 1;
-	bool from_string(std::string_view value);
-	std::string to_string() const;
-	void to_string(char* to) const noexcept;
-
-	// TODO: provide compatibilitiy with some well-known system
-};
-
-std::ostream& operator<<(std::ostream& os, const datetime& value);
-
-}
-
-#include "mysql/impl/datetime_impl.hpp"
-
-#endif /* INCLUDE_MYSQL_DATETIME_HPP_ */
diff --git a/include/mysql/impl/connection_impl.hpp b/include/mysql/impl/connection_impl.hpp
index 2608a375..67fed963 100644
--- a/include/mysql/impl/connection_impl.hpp
+++ b/include/mysql/impl/connection_impl.hpp
@@ -2,8 +2,10 @@
 #define INCLUDE_MYSQL_IMPL_CONNECTION_IMPL_HPP_
 
 #include "mysql/impl/handshake.hpp"
+#include "mysql/impl/query.hpp"
 #include <boost/asio/buffer.hpp>
 
+// Handshake
 template <typename Stream, typename Allocator>
 void mysql::connection<Stream, Allocator>::handshake(
 	const connection_params& params,
@@ -40,6 +42,31 @@ mysql::connection<Stream, Allocator>::async_handshake(
 	);
 }
 
+// Query
+template <typename Stream, typename Allocator>
+mysql::resultset<typename mysql::connection<Stream, Allocator>::channel_type, Allocator>
+mysql::connection<Stream, Allocator>::query(
+	std::string_view query_string,
+	error_code& err
+)
+{
+	resultset<channel_type, Allocator> res;
+	detail::execute_query(channel_, query_string, res, err);
+	return res;
+}
+
+template <typename Stream, typename Allocator>
+mysql::resultset<typename mysql::connection<Stream, Allocator>::channel_type, Allocator>
+mysql::connection<Stream, Allocator>::query(
+	std::string_view query_string
+)
+{
+	resultset<channel_type, Allocator> res;
+	error_code err;
+	detail::execute_query(channel_, query_string, res, err);
+	detail::check_error_code(err);
+	return res;
+}
 
 
 #endif /* INCLUDE_MYSQL_IMPL_CONNECTION_IMPL_HPP_ */
diff --git a/include/mysql/impl/datetime_impl.hpp b/include/mysql/impl/datetime_impl.hpp
deleted file mode 100644
index fd0fd56c..00000000
--- a/include/mysql/impl/datetime_impl.hpp
+++ /dev/null
@@ -1,104 +0,0 @@
-#ifndef INCLUDE_MYSQL_IMPL_DATETIME_IMPL_HPP_
-#define INCLUDE_MYSQL_IMPL_DATETIME_IMPL_HPP_
-
-#include <cstdio>
-#include <cstring>
-#include <cassert>
-
-constexpr mysql::datetime::datetime(
-	std::uint16_t years,
-	std::uint8_t month,
-	std::uint8_t day,
-	std::uint8_t hour,
-	std::uint8_t min,
-	std::uint8_t seconds,
-	std::uint32_t micro
-) noexcept :
-	year_(years),
-	month_(month),
-	day_(day),
-	hour_(hour),
-	minute_(min),
-	second_(seconds),
-	microsecond_(micro)
-{
-}
-
-constexpr bool mysql::datetime::operator==(
-	const datetime& rhs
-) const noexcept
-{
-	return year_ == rhs.year_ &&
-		   month_ == rhs.month_ &&
-		   day_ == rhs.day_ &&
-		   hour_ == rhs.hour_ &&
-		   minute_ == rhs.minute_ &&
-		   second_ == rhs.second_ &&
-		   microsecond_ == rhs.microsecond_;
-}
-
-bool mysql::datetime::from_string(
-	std::string_view value
-)
-{
-	if (value.size() > max_string_size - 1) return false;
-	char buffer [max_string_size] {};
-	memcpy(buffer, value.data(), value.size());
-	unsigned years {}, months {}, days {}, hours {}, mins {}, seconds {}, microseconds {};
-	int elms_parsed = sscanf(buffer, "%u-%u-%u %u:%u:%u.%u",
-			&years, &months, &days, &hours, &mins, &seconds, &microseconds);
-	if (elms_parsed < 3 ||
-	    months > 12 ||
-		days > 31 ||
-		hours > 24 ||
-		mins > 60 ||
-		seconds > 60 ||
-		microseconds > 1000000)
-	{
-		return false;
-	}
-	year_ = years;
-	month_ = months;
-	day_ = days;
-	hour_ = hours;
-	minute_ = mins;
-	second_ = seconds;
-	microsecond_ = microseconds;
-	return true;
-}
-
-inline void mysql::datetime::to_string(
-	char* to
-) const noexcept
-{
-	unsigned years = std::min(static_cast<unsigned>(year()), 9999u);
-	unsigned months = std::min(month(), std::uint8_t(12));
-	unsigned days = std::min(day(), std::uint8_t(31));
-	unsigned hrs = std::min(hour(), std::uint8_t(24));
-	unsigned mins = std::min(minute(), std::uint8_t(60));
-	unsigned secs = std::min(second(), std::uint8_t(60));
-	unsigned micros = std::min(microsecond(), 999999u);
-	int result = snprintf(to, max_string_size, "%4u-%2u-%2u %2u:%2u:%2u.%6u",
-			years, months, days, hrs, mins, secs, micros);
-	assert(result == (max_string_size - 1));
-}
-
-inline std::string mysql::datetime::to_string() const
-{
-	std::string res (max_string_size, 0);
-	to_string(res.data());
-	res.pop_back();
-	return res;
-}
-
-inline std::ostream& mysql::operator<<(
-	std::ostream& os,
-	const datetime& value
-)
-{
-	char buffer [datetime::max_string_size];
-	value.to_string(buffer);
-	return os << buffer;
-}
-
-#endif /* INCLUDE_MYSQL_IMPL_DATETIME_IMPL_HPP_ */
diff --git a/include/mysql/impl/deserialize_row.hpp b/include/mysql/impl/deserialize_row.hpp
new file mode 100644
index 00000000..5c7a4479
--- /dev/null
+++ b/include/mysql/impl/deserialize_row.hpp
@@ -0,0 +1,34 @@
+#ifndef INCLUDE_MYSQL_IMPL_DESERIALIZE_ROW_HPP_
+#define INCLUDE_MYSQL_IMPL_DESERIALIZE_ROW_HPP_
+
+#include "mysql/error.hpp"
+#include "mysql/value.hpp"
+#include "mysql/metadata.hpp"
+#include "mysql/value.hpp"
+#include "mysql/impl/basic_serialization.hpp"
+#include <vector>
+
+namespace mysql
+{
+namespace detail
+{
+
+inline Error deserialize_text_value(
+	std::string_view from,
+	const field_metadata& meta,
+	value& output
+);
+
+template <typename Allocator>
+error_code deserialize_text_row(
+	DeserializationContext& ctx,
+	const resultset_metadata<Allocator>& meta,
+	std::vector<value>& output
+);
+
+}
+}
+
+#include "mysql/impl/deserialize_row_impl.hpp"
+
+#endif /* INCLUDE_MYSQL_IMPL_DESERIALIZE_ROW_HPP_ */
diff --git a/include/mysql/impl/deserialize_row_impl.hpp b/include/mysql/impl/deserialize_row_impl.hpp
new file mode 100644
index 00000000..449bc939
--- /dev/null
+++ b/include/mysql/impl/deserialize_row_impl.hpp
@@ -0,0 +1,218 @@
+#ifndef INCLUDE_MYSQL_IMPL_DESERIALIZE_ROW_IMPL_HPP_
+#define INCLUDE_MYSQL_IMPL_DESERIALIZE_ROW_IMPL_HPP_
+
+#include <cstdlib>
+#include <cmath>
+#include <boost/lexical_cast/try_lexical_convert.hpp>
+#include <date/date.h>
+
+namespace mysql
+{
+namespace detail
+{
+
+inline Error deserialize_text_value_impl(
+	std::string_view from,
+	date& to
+)
+{
+	constexpr std::size_t size = 4 + 2 + 2 + 2; // year, month, day, separators
+	if (from.size() != size) return Error::protocol_value_error;
+	unsigned year, month, day;
+	char buffer [size + 1] {};
+	memcpy(buffer, from.data(), from.size());
+	int parsed = sscanf(buffer, "%4u-%2u-%2u", &year, &month, &day);
+	if (parsed != 3) return Error::protocol_value_error;
+	::date::year_month_day result (::date::year(year)/::date::month(month)/::date::day(day));
+	if (!result.ok()) return Error::protocol_value_error;
+	if (result > max_date || result < min_date) return Error::protocol_value_error;
+	to = result;
+	return Error::ok;
+}
+
+inline Error deserialize_text_value_impl(
+	std::string_view from,
+	time& to,
+	unsigned decimals
+)
+{
+	// size check
+	constexpr std::size_t min_size = 2 + 2 + 2 + 2; // hours, mins, seconds, no micros
+	constexpr std::size_t max_size = min_size + 1 + 7; // hour extra character and micros
+	decimals = std::min(decimals, 6u);
+	if (from.size() < min_size || from.size() > max_size) return Error::protocol_value_error;
+
+	// Parse it
+	int hours;
+	unsigned minutes, seconds, micros = 0;
+	char buffer [max_size + 1] {};
+	memcpy(buffer, from.data(), from.size());
+	int parsed = decimals ? sscanf(buffer, "%3d:%2u:%2u.%6u", &hours, &minutes, &seconds, &micros) :
+			                sscanf(buffer, "%3d:%2u:%2u", &hours, &minutes, &seconds);
+	if ((decimals && parsed != 4) || parsed != 3) return Error::protocol_value_error;
+	micros *= std::pow(10, 6 - decimals);
+	bool is_negative = hours < 0;
+	hours = std::abs(hours);
+
+	// Sum it
+	auto res = std::chrono::hours(hours) + std::chrono::minutes(minutes) +
+			   std::chrono::seconds(seconds) + std::chrono::microseconds(micros);
+	if (is_negative)
+	{
+		res = -res;
+	}
+
+	// Range check
+	if (res > max_time || res < min_time) return Error::protocol_value_error;
+
+	to = res;
+	return Error::ok;
+}
+
+inline Error deserialize_text_value_impl(
+	std::string_view from,
+	datetime& to,
+	unsigned decimals
+)
+{
+	// Length check
+	constexpr std::size_t min_size = 4 + 5*2 + 5; // year, month, day, hour, minute, seconds, separators
+	constexpr std::size_t max_size = min_size + 7;
+	decimals = std::min(decimals, 6u);
+	std::size_t expected_size = min_size + (decimals ? decimals + 1 : 0);
+	if (from.size() != expected_size) return Error::protocol_value_error;
+
+	// Date part
+	date dt;
+	auto err = deserialize_text_value_impl(from.substr(0, 10), dt);
+	if (err != Error::ok) return err;
+
+	// Time of day part
+	time time_of_day;
+	err = deserialize_text_value_impl(from.substr(11), time_of_day, decimals);
+	if (err != Error::ok) return err;
+	constexpr auto max_time_of_day = std::chrono::hours(24) - std::chrono::microseconds(1);
+	if (time_of_day < std::chrono::seconds(0) || time_of_day > max_time_of_day) return Error::protocol_value_error;
+
+	// Sum it up
+	to = dt + time_of_day;
+	return Error::ok;
+}
+
+
+template <typename T>
+std::enable_if_t<std::is_arithmetic_v<T>, Error>
+deserialize_text_value_impl(std::string_view from, T& to)
+{
+	bool ok = boost::conversion::try_lexical_convert(from.data(), from.size(), to);
+	return ok ? Error::ok : Error::protocol_value_error;
+}
+
+Error deserialize_text_value_impl(std::string_view from, std::string_view& to)
+{
+	to = from;
+	return Error::ok;
+}
+
+Error deserialize_text_value_impl(std::string_view from, std::nullptr_t& to)
+{
+	return Error::ok;
+}
+
+template <typename T, typename... Args>
+Error deserialize_text_value_to_variant(std::string_view from, value& to, Args&&... args)
+{
+	T value;
+	auto err = deserialize_text_value_impl(from, value, std::forward<Args>(args)...);
+	if (err == Error::ok)
+	{
+		to = value;
+	}
+	return err;
+}
+
+}
+}
+
+inline mysql::Error mysql::detail::deserialize_text_value(
+	std::string_view from,
+	const field_metadata& meta,
+	value& output
+)
+{
+	switch (meta.type())
+	{
+    case field_type::decimal:
+    case field_type::varchar:
+    case field_type::bit:
+    case field_type::newdecimal:
+    case field_type::enum_:
+    case field_type::set:
+    case field_type::tiny_blob:
+    case field_type::medium_blob:
+    case field_type::long_blob:
+    case field_type::blob:
+    case field_type::var_string:
+    case field_type::string:
+    case field_type::geometry:
+    	return deserialize_text_value_to_variant<std::string_view>(from, output);
+    case field_type::tiny:
+    	return meta.is_unsigned() ?
+    			deserialize_text_value_to_variant<std::uint8_t>(from, output) :
+				deserialize_text_value_to_variant<std::int8_t>(from, output);
+    case field_type::short_:
+    	return meta.is_unsigned() ?
+    			deserialize_text_value_to_variant<std::uint16_t>(from, output) :
+				deserialize_text_value_to_variant<std::int16_t>(from, output);
+    case field_type::int24:
+    case field_type::long_:
+    case field_type::year:
+    	return meta.is_unsigned() ?
+    			deserialize_text_value_to_variant<std::uint32_t>(from, output) :
+				deserialize_text_value_to_variant<std::int32_t>(from, output);
+    case field_type::longlong:
+    	return meta.is_unsigned() ?
+    			deserialize_text_value_to_variant<std::uint64_t>(from, output) :
+				deserialize_text_value_to_variant<std::int64_t>(from, output);
+    case field_type::float_:
+    	return deserialize_text_value_to_variant<float>(from, output);
+    case field_type::double_:
+    	return deserialize_text_value_to_variant<double>(from, output);
+    case field_type::null:
+    	return deserialize_text_value_to_variant<nullptr_t>(from, output);
+    case field_type::timestamp:
+    case field_type::datetime:
+    	return deserialize_text_value_to_variant<datetime>(from, output, meta.decimals());
+    case field_type::date:
+    	return deserialize_text_value_to_variant<date>(from, output);
+    case field_type::time:
+    	return deserialize_text_value_to_variant<time>(from, output, meta.decimals());
+    default:
+    	return Error::protocol_value_error;
+	}
+}
+
+
+template <typename Allocator>
+mysql::error_code mysql::detail::deserialize_text_row(
+	DeserializationContext& ctx,
+	const resultset_metadata<Allocator>& meta,
+	std::vector<value>& output
+)
+{
+	output.resize(meta.fields().size());
+	for (std::vector<value>::size_type i = 0; i < meta.fields().size(); ++i)
+	{
+		string_lenenc value_str;
+		Error err = deserialize(value_str, ctx);
+		if (err) return make_error_code(err);
+		err = parse_text_value(value_str.value, meta.fields()[i], output[i]);
+		if (err) return make_error_code(err);
+	}
+	if (!ctx.empty()) return make_error_code(Error::extra_bytes);
+	return error_code();
+}
+
+
+
+#endif /* INCLUDE_MYSQL_IMPL_DESERIALIZE_ROW_IMPL_HPP_ */
diff --git a/include/mysql/impl/query.hpp b/include/mysql/impl/query.hpp
new file mode 100644
index 00000000..74863878
--- /dev/null
+++ b/include/mysql/impl/query.hpp
@@ -0,0 +1,43 @@
+#ifndef INCLUDE_MYSQL_IMPL_QUERY_HPP_
+#define INCLUDE_MYSQL_IMPL_QUERY_HPP_
+
+#include "mysql/resultset.hpp"
+#include "mysql/impl/capabilities.hpp"
+#include <string_view>
+
+namespace mysql
+{
+namespace detail
+{
+
+enum class fetch_result
+{
+	error,
+	row,
+	eof
+};
+
+template <typename ChannelType, typename Allocator>
+void execute_query(
+	ChannelType& channel,
+	std::string_view query,
+	resultset<ChannelType, Allocator>& output,
+	error_code& err
+);
+
+template <typename ChannelType, typename Allocator>
+fetch_result fetch_text_row(
+	ChannelType& channel,
+	const resultset_metadata<Allocator>& meta,
+	bytestring<Allocator>& buffer,
+	std::vector<value>& output_values,
+	msgs::ok_packet& output_ok_packet,
+	error_code& err
+);
+
+}
+}
+
+#include "mysql/impl/query_impl.hpp"
+
+#endif /* INCLUDE_MYSQL_IMPL_QUERY_HPP_ */
diff --git a/include/mysql/impl/query_impl.hpp b/include/mysql/impl/query_impl.hpp
new file mode 100644
index 00000000..2c8fea2c
--- /dev/null
+++ b/include/mysql/impl/query_impl.hpp
@@ -0,0 +1,133 @@
+#ifndef INCLUDE_MYSQL_IMPL_QUERY_IMPL_HPP_
+#define INCLUDE_MYSQL_IMPL_QUERY_IMPL_HPP_
+
+#include "mysql/impl/messages.hpp"
+#include "mysql/impl/basic_serialization.hpp"
+#include "mysql/impl/deserialize_row.hpp"
+
+template <typename ChannelType, typename Allocator>
+void mysql::detail::execute_query(
+	ChannelType& channel,
+	std::string_view query,
+	resultset<ChannelType, Allocator>& output,
+	error_code& err
+)
+{
+	// Compose a com_query message
+	msgs::com_query query_msg;
+	query_msg.query.value = query;
+
+	// Serialize it
+	capabilities caps = channel.current_capabilities();
+	bytestring<Allocator> buffer;
+	serialize_message(query_msg, caps, buffer);
+
+	// Send it
+	channel.write(boost::asio::buffer(buffer), err);
+	if (err) return;
+
+	// Read the response
+	channel.read(buffer, err);
+	if (err) return;
+
+	// Response may be: ok_packet, err_packet, local infile request (TODO)
+	// If it is none of this, then the message type itself is the beginning of
+	// a length-encoded int containing the field count
+	DeserializationContext ctx (boost::asio::buffer(buffer), caps);
+	std::uint8_t msg_type;
+	err = deserialize_message_type(msg_type, ctx);
+	if (err) return;
+	if (msg_type == ok_packet_header)
+	{
+		msgs::ok_packet ok_packet;
+		err = deserialize_message(ok_packet, ctx);
+		if (err) return;
+		output = resultset<ChannelType, Allocator>(channel, std::move(buffer), ok_packet);
+		err.clear();
+		return;
+	}
+	else if (msg_type == error_packet_header)
+	{
+		err = process_error_packet(ctx);
+		return;
+	}
+
+	// Resultset with metadata. First packet is an int_lenenc with
+	// the number of field definitions to expect. Message type is part
+	// of this packet, so we must rewind the context
+	ctx.set_first(buffer.data());
+	int_lenenc num_fields;
+	err = deserialize_message(num_fields, ctx);
+	if (err) return;
+
+	std::vector<owning_field_metadata<Allocator>> fields;
+	fields.reserve(num_fields.value);
+
+	// Read all of the field definitions
+	for (std::uint64_t i = 0; i < num_fields.value; ++i)
+	{
+		// Read the field definition packet
+		bytestring<Allocator> field_definition_buffer;
+		channel.read(field_definition_buffer, err);
+		if (err) return;
+
+		// Deserialize the message
+		msgs::column_definition field_definition;
+		ctx = DeserializationContext(boost::asio::buffer(field_definition_buffer), caps);
+		err = deserialize_message(field_definition, field_definition_buffer);
+		if (err) return;
+
+		// Add it to our array
+		fields.emplace_back(std::move(field_definition_buffer), field_definition);
+	}
+
+	// No EOF packet is expected here, as we require deprecate EOF capabilities
+	output = resultset<ChannelType, Allocator>(channel, std::move(fields));
+	err.clear();
+}
+
+template <typename ChannelType, typename Allocator>
+mysql::detail::fetch_result mysql::detail::fetch_text_row(
+	ChannelType& channel,
+	const resultset_metadata<Allocator>& meta,
+	bytestring<Allocator>& buffer,
+	std::vector<value>& output_values,
+	msgs::ok_packet& output_ok_packet,
+	error_code& err
+)
+{
+	// Read a packet
+	channel.read(buffer, err);
+	if (err) return fetch_result::error;
+
+	// Message type: row, error or eof?
+	std::uint8_t msg_type;
+	DeserializationContext ctx (boost::asio::buffer(buffer), channel.current_capabilities());
+	err = deserialize_message_type(msg_type, ctx);
+	if (err) return fetch_result::error;
+	if (msg_type == eof_packet_header)
+	{
+		// end of resultset
+		err = deserialize_message(output_ok_packet, ctx);
+		if (err) return fetch_result::error;
+		return fetch_result::eof;
+	}
+	else if (msg_type == error_packet_header)
+	{
+		// An error occurred during the generation of the rows
+		err = process_error_packet(ctx);
+		return fetch_result::error;
+	}
+	else
+	{
+		// An actual row
+		err = deserialize_text_row(ctx, meta, output_values);
+		if (err) return fetch_result::error;
+		return fetch_result::row;
+	}
+}
+
+
+
+
+#endif /* INCLUDE_MYSQL_IMPL_QUERY_IMPL_HPP_ */
diff --git a/include/mysql/impl/resultset_impl.hpp b/include/mysql/impl/resultset_impl.hpp
new file mode 100644
index 00000000..17bfab53
--- /dev/null
+++ b/include/mysql/impl/resultset_impl.hpp
@@ -0,0 +1,28 @@
+#ifndef INCLUDE_MYSQL_IMPL_RESULTSET_IMPL_HPP_
+#define INCLUDE_MYSQL_IMPL_RESULTSET_IMPL_HPP_
+
+#include "mysql/impl/query.hpp"
+#include <cassert>
+
+template <typename ChannelType, typename Allocator>
+const mysql::row<Allocator>* mysql::resultset<ChannelType, Allocator>::fetch_one(
+	error_code& err
+)
+{
+	assert(channel_);
+	assert(!complete());
+	auto result = detail::fetch_text_row(
+		*channel_,
+		fields_,
+		buffer_,
+		current_row_.values(),
+		ok_packet_,
+		err
+	);
+	eof_received_ = result == detail::fetch_result::eof;
+	return result == detail::fetch_result::row ? &current_row_ : nullptr;
+}
+
+
+
+#endif /* INCLUDE_MYSQL_IMPL_RESULTSET_IMPL_HPP_ */
diff --git a/include/mysql/metadata.hpp b/include/mysql/metadata.hpp
index b74af77d..f2ae01e3 100644
--- a/include/mysql/metadata.hpp
+++ b/include/mysql/metadata.hpp
@@ -7,22 +7,14 @@
 namespace mysql
 {
 
-template <typename Allocator>
 class field_metadata
 {
-	detail::bytestring<Allocator> msg_buffer_;
-	detail::msgs::column_definition msg_; // holds pointers into the message buffer
+	detail::msgs::column_definition msg_;
 
 	bool flag_set(std::uint16_t flag) const noexcept { return msg_.flags.value & flag; }
 public:
 	field_metadata() = default;
-	field_metadata(detail::bytestring<Allocator>&& buffer, const detail::msgs::column_definition& msg):
-		msg_buffer_(std::move(buffer)), msg_(msg) {};
-	field_metadata(const field_metadata&) = delete;
-	field_metadata(field_metadata&&) = default;
-	field_metadata& operator=(const field_metadata&) = delete;
-	field_metadata& operator=(field_metadata&&) = default;
-	~field_metadata() = default;
+	field_metadata(const detail::msgs::column_definition& msg) noexcept: msg_(msg) {};
 
 	std::string_view database() const noexcept { return msg_.schema.value; }
 	std::string_view table() const noexcept { return msg_.table.value; }
@@ -49,10 +41,27 @@ public:
 	bool is_set_to_now_on_update() const noexcept { return flag_set(detail::column_flags::on_update_now); }
 };
 
+template <typename Allocator>
+class owning_field_metadata : public field_metadata
+{
+	detail::bytestring<Allocator> msg_buffer_;
+
+	bool flag_set(std::uint16_t flag) const noexcept { return msg_.flags.value & flag; }
+public:
+	owning_field_metadata() = default;
+	owning_field_metadata(detail::bytestring<Allocator>&& buffer, const detail::msgs::column_definition& msg):
+		field_metadata(msg), msg_buffer_(std::move(buffer)) {};
+	owning_field_metadata(const owning_field_metadata&) = delete;
+	owning_field_metadata(owning_field_metadata&&) = default;
+	owning_field_metadata& operator=(const owning_field_metadata&) = delete;
+	owning_field_metadata& operator=(owning_field_metadata&&) = default;
+	~owning_field_metadata() = default;
+};
+
 template <typename Allocator>
 class resultset_metadata
 {
-	std::vector<field_metadata<Allocator>> fields_;
+	std::vector<owning_field_metadata<Allocator>> fields_;
 public:
 	const auto& fields() const noexcept { return fields_; }
 };
diff --git a/include/mysql/resultset.hpp b/include/mysql/resultset.hpp
new file mode 100644
index 00000000..5e92d107
--- /dev/null
+++ b/include/mysql/resultset.hpp
@@ -0,0 +1,51 @@
+#ifndef INCLUDE_MYSQL_RESULTSET_HPP_
+#define INCLUDE_MYSQL_RESULTSET_HPP_
+
+#include "mysql/row.hpp"
+#include "mysql/metadata.hpp"
+#include "mysql/impl/messages.hpp"
+#include <cassert>
+
+namespace mysql
+{
+
+// TODO: provide a wrapper so ChannelType does not have to be specified directly
+template <typename ChannelType, typename Allocator>
+class resultset
+{
+	ChannelType* channel_;
+	resultset_metadata<Allocator> fields_;
+	row<Allocator> current_row_;
+	detail::bytestring<Allocator> buffer_;
+	detail::msgs::ok_packet ok_packet_;
+	bool eof_received_ {false};
+public:
+	resultset(): channel_(nullptr) {};
+	resultset(ChannelType& channel, resultset_metadata<Allocator>&& fields):
+		channel_(&channel), fields_(std::move(fields)) {};
+	resultset(ChannelType& channel, detail::bytestring<Allocator>&& buffer, const detail::msgs::ok_packet& ok_pack):
+		channel_(&channel), buffer_(std::move(buffer)), ok_packet_(ok_pack), eof_received_(true) {};
+
+	const row<Allocator>* fetch_one(error_code& err);
+	const row<Allocator>* fetch_one();
+	std::vector<owning_row<Allocator>> fetch_many(std::size_t count, error_code& err);
+	std::vector<owning_row<Allocator>> fetch_many(std::size_t count);
+	std::vector<owning_row<Allocator>> fetch_all(error_code& err);
+	std::vector<owning_row<Allocator>> fetch_all();
+
+	// Is the read of the resultset complete? Pre-condition to any of the functions
+	// accessing the ok_packet
+	bool complete() const noexcept { return eof_received_; }
+	const auto& fields() const noexcept { return fields_; }
+	std::uint64_t affected_rows() const noexcept { assert(complete()); return ok_packet_.affected_rows.value; }
+	std::uint64_t last_insert_id() const noexcept { assert(complete()); return ok_packet_.last_insert_id.value; }
+	unsigned warning_count() const noexcept { assert(complete()); return ok_packet_.warnings.value; }
+	std::string_view info() const noexcept { assert(complete()); return ok_packet_.info.value; }
+	// TODO: status flags accessors
+};
+
+}
+
+#include "mysql/impl/resultset_impl.hpp"
+
+#endif /* INCLUDE_MYSQL_RESULTSET_HPP_ */
diff --git a/include/mysql/row.hpp b/include/mysql/row.hpp
index edab92f5..4f6d5cb4 100644
--- a/include/mysql/row.hpp
+++ b/include/mysql/row.hpp
@@ -11,26 +11,32 @@ namespace mysql
 template <typename Allocator>
 class row
 {
-	detail::bytestring<Allocator> buffer_;
 	std::vector<value> values_;
 	const resultset_metadata<Allocator>* metadata_;
 public:
 	row(): metadata_(nullptr) {};
-	row(detail::bytestring<Allocator>&& buffer, std::vector<value>&& values,
-			const resultset_metadata<Allocator>& meta):
-		buffer_(std::move(buffer)), values_(std::move(values)), metadata_(&meta) {};
-	row(const row&) = delete;
-	row(row&&) = default;
-	row& operator=(const row&) = delete;
-	row& operator=(row&&) = default;
-	~row() = default;
+	row(std::vector<value>&& values, const resultset_metadata<Allocator>& meta):
+		values_(std::move(values)), metadata_(&meta) {};
 
 	const std::vector<value>& values() const noexcept { return values_; }
+	std::vector<value>& values() noexcept { return values_; }
 	const auto& metadata() const noexcept { return *metadata_; }
+};
 
-	// TODO: can we make these private? accessed by resultset
-	auto& buffer() { return buffer_; }
-	auto& values() { return values_; }
+template <typename Allocator>
+class owning_row : public row<Allocator>
+{
+	detail::bytestring<Allocator> buffer_;
+public:
+	owning_row() = default;
+	owning_row(std::vector<value>&& values, const resultset_metadata<Allocator>& meta,
+			detail::bytestring<Allocator>&& buffer) :
+			row<Allocator>(std::move(values), meta), buffer_(std::move(buffer)) {};
+	owning_row(const owning_row&) = delete;
+	owning_row(owning_row&&) = default;
+	owning_row& operator=(const owning_row&) = delete;
+	owning_row& operator=(owning_row&&) = default;
+	~owning_row() = default;
 };
 
 }
diff --git a/include/mysql/value.hpp b/include/mysql/value.hpp
index e97092e6..f395268b 100644
--- a/include/mysql/value.hpp
+++ b/include/mysql/value.hpp
@@ -4,30 +4,15 @@
 #include <variant>
 #include <cstdint>
 #include <string_view>
+#include <date/date.h>
+#include <chrono>
 
 namespace mysql
 {
 
-// TODO: decide a better interface for these types
-struct datetime // DATETIME, DATE, TIMESTAMP
-{
-	std::uint16_t year;
-	std::uint8_t month;
-	std::uint8_t day;
-	std::uint8_t hour;
-	std::uint8_t minutes;
-	std::uint32_t microseconds;
-};
-
-struct time // TIME
-{
-	bool is_negative;
-	std::uint32_t days;
-	std::uint8_t hours;
-	std::uint8_t minutes;
-	std::uint8_t seconds;
-	std::uint32_t microseconds;
-};
+using date = ::date::sys_days;
+using datetime = ::date::sys_time<std::chrono::microseconds>;
+using time = std::chrono::microseconds;
 
 /**
  * field_type::decimal: string_view
@@ -42,16 +27,16 @@ struct time // TIME
  * field_type::var_string: string_view
  * field_type::geometry: string_view
  * field_type::string: string_view
- * field_type::tiny: uint8_t/int8_t
- * field_type::short: uint16_t/int8_t
+ * field_type::tiny: (u)int8_t
+ * field_type::short: (u)int16_t
  * field_type::year: uint16_t
- * field_type::int24: uint32_t/int8_t
- * field_type::long_: uint32_t/int8_t
- * field_type::longlong: uint64_t/int8_t
+ * field_type::int24: (u)int32_t
+ * field_type::long_: (u)int32_t
+ * field_type::longlong: (u)int64_t
  * field_type::float_: float
  * field_type::double_: double
  * field_type::timestamp: datetime
- * field_type::date: datetime
+ * field_type::date: date
  * field_type::datetime: datetime
  * field_type::time: time
  * field_type::null: nullptr_t
@@ -68,6 +53,7 @@ using value = std::variant<
 	std::string_view,
 	float,
 	double,
+	date,
 	datetime,
 	time,
 	std::nullptr_t
@@ -75,6 +61,27 @@ using value = std::variant<
 
 }
 
+// Range checks
+namespace mysql
+{
+namespace detail
+{
 
+constexpr date min_date = ::date::day(1)/::date::January/::date::year(1000);
+constexpr date max_date = ::date::day(31)/::date::December/::date::year(9999);
+constexpr datetime min_datetime = min_date;
+constexpr datetime max_datetime = max_date + std::chrono::hours(24) - std::chrono::microseconds(1);
+constexpr time min_time = -std::chrono::hours(839);
+constexpr time max_time = std::chrono::hours(839);
+
+static_assert(date::min() <= min_date);
+static_assert(date::max() >= max_date);
+static_assert(datetime::min() <= min_datetime);
+static_assert(datetime::max() >= max_datetime);
+static_assert(time::min() <= min_time);
+static_assert(time::max() >= max_time);
+
+}
+}
 
 #endif /* INCLUDE_MYSQL_VALUE_HPP_ */
diff --git a/test/datetime.cpp b/test/datetime.cpp
deleted file mode 100644
index 9f8c27e9..00000000
--- a/test/datetime.cpp
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * datetime.cpp
- *
- *  Created on: Nov 2, 2019
- *      Author: ruben
- */
-
-#include "mysql/datetime.hpp"
-#include <gtest/gtest.h>
-
-using namespace testing;
-using namespace mysql;
-
-TEST(Datetime, DefaultConstructor_Trivial_AllZeros)
-{
-	datetime dt;
-	EXPECT_EQ(dt.year(), 0);
-	EXPECT_EQ(dt.month(), 0);
-	EXPECT_EQ(dt.day(), 0);
-	EXPECT_EQ(dt.hour(), 0);
-	EXPECT_EQ(dt.minute(), 0);
-	EXPECT_EQ(dt.second(), 0);
-	EXPECT_EQ(dt.microsecond(), 0);
-}
-
-TEST(Datetime, InitializationConstructor_Trivial_SetsAllMembers)
-{
-	datetime dt (2010, 10, 9, 22, 50, 55, 2090);
-	EXPECT_EQ(dt.year(), 2010);
-	EXPECT_EQ(dt.month(), 10);
-	EXPECT_EQ(dt.day(), 9);
-	EXPECT_EQ(dt.hour(), 22);
-	EXPECT_EQ(dt.minute(), 50);
-	EXPECT_EQ(dt.second(), 55);
-	EXPECT_EQ(dt.microsecond(), 2090);
-}
-
-TEST(Datetime, InitializationConstructor_DefaultArgs_SetsThemToZero)
-{
-	datetime dt (2010, 10, 9);
-	EXPECT_EQ(dt.year(), 2010);
-	EXPECT_EQ(dt.month(), 10);
-	EXPECT_EQ(dt.day(), 9);
-	EXPECT_EQ(dt.hour(), 0);
-	EXPECT_EQ(dt.minute(), 0);
-	EXPECT_EQ(dt.second(), 0);
-	EXPECT_EQ(dt.microsecond(), 0);
-}
-
-TEST(Datetime, Setters_Trivial_SetRelevantMember)
-{
-	datetime dt;
-	dt.set_year(2010);
-	dt.set_month(10);
-	dt.set_day(9);
-	dt.set_hour(22);
-	dt.set_minute(50);
-	dt.set_second(55);
-	dt.set_microsecond(2090);
-	EXPECT_EQ(dt.year(), 2010);
-	EXPECT_EQ(dt.month(), 10);
-	EXPECT_EQ(dt.day(), 9);
-	EXPECT_EQ(dt.hour(), 22);
-	EXPECT_EQ(dt.minute(), 50);
-	EXPECT_EQ(dt.second(), 55);
-	EXPECT_EQ(dt.microsecond(), 2090);
-}
-
-TEST(Datetime, OperatorEquals_AllMembersEqual_ReturnsTrue)
-{
-	EXPECT_EQ(datetime(), datetime());
-	EXPECT_EQ((datetime(2010, 9, 10)), (datetime(2010, 9, 10)));
-	EXPECT_EQ((datetime(2010, 9, 10, 22, 59, 60, 9999)), (datetime(2010, 9, 10, 22, 59, 60, 9999)));
-}
-
-TEST(Datetime, OperatorEquals_AnyNonEqualMember_ReturnsFalse)
-{
-	EXPECT_FALSE((datetime(2010, 0, 0)) == datetime());
-	EXPECT_FALSE((datetime(2010, 9, 10)) == (datetime(2010, 10, 10)));
-	EXPECT_FALSE((datetime(2010, 9, 10, 22, 59, 60, 9999)) == (datetime(2011, 9, 10, 22, 59, 60, 9999)));
-	EXPECT_FALSE((datetime(2010, 9, 10, 22, 59, 60, 9999)) == (datetime(2010, 8, 10, 22, 59, 60, 9999)));
-	EXPECT_FALSE((datetime(2010, 9, 10, 22, 59, 60, 9999)) == (datetime(2010, 9, 11, 22, 59, 60, 9999)));
-	EXPECT_FALSE((datetime(2010, 9, 10, 22, 59, 60, 9999)) == (datetime(2010, 9, 10, 23, 59, 60, 9999)));
-	EXPECT_FALSE((datetime(2010, 9, 10, 22, 59, 60, 9999)) == (datetime(2010, 9, 10, 22, 55, 60, 9999)));
-	EXPECT_FALSE((datetime(2010, 9, 10, 22, 59, 60, 9999)) == (datetime(2010, 9, 10, 22, 59, 59, 9999)));
-	EXPECT_FALSE((datetime(2010, 9, 10, 22, 59, 60, 9999)) == (datetime(2010, 9, 10, 22, 59, 60, 88)));
-}
-
-TEST(Datetime, OperatorNotEquals_Trivial_ReturnsNotEquals)
-{
-	EXPECT_TRUE(datetime() != datetime(2010, 0, 0));
-	EXPECT_FALSE(datetime(2010, 0, 0) != datetime(2010, 0, 0));
-}
-
-
-
diff --git a/test/handshake.cpp b/test/handshake.cpp
new file mode 100644
index 00000000..2fcab205
--- /dev/null
+++ b/test/handshake.cpp
@@ -0,0 +1,33 @@
+/*
+ * handshake.cpp
+ *
+ *  Created on: Oct 27, 2019
+ *      Author: ruben
+ */
+
+#include <gtest/gtest.h>
+#include <gmock/gmock.h>
+#include "mysql/impl/handshake.hpp"
+
+using namespace testing;
+using namespace mysql;
+using namespace detail;
+
+namespace
+{
+
+class MockChannel
+{
+public:
+	MOCK_METHOD2(read_impl, void(std::vector<std::uint8_t>&, error_code&));
+	MOCK_METHOD2(write, void(boost::asio::const_buffer, error_code&));
+
+	template <typename Allocator>
+	void read(bytestring<Allocator>& buffer, error_code& errc)
+	{
+		read_impl(buffer, errc);
+	}
+};
+
+}
+
diff --git a/test/integration/db_setup.sh b/test/integration/db_setup.sh
new file mode 100755
index 00000000..51e3100f
--- /dev/null
+++ b/test/integration/db_setup.sh
@@ -0,0 +1,3 @@
+#!/bin/bash
+
+mysql -u root -p < db_setup.sql
diff --git a/test/integration/db_setup.sql b/test/integration/db_setup.sql
index 6200c17b..b9b8d1fc 100644
--- a/test/integration/db_setup.sql
+++ b/test/integration/db_setup.sql
@@ -6,14 +6,21 @@ USE awesome;
 -- Tables
 CREATE TABLE test_table (
     id INT AUTO_INCREMENT PRIMARY KEY,
+    field_decimal DECIMAL (6,2),
     field_varchar VARCHAR(255) NOT NULL,
+    field_bit BIT(8),
     field_float FLOAT(20),
-    field_date DATE,
-    field_tiny TINYINT NOT NULL,
+    field_tiny TINYINT,
     field_text TEXT,
-    field_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+    field_date DATE,
+    field_datetime DATETIME (3),
+    field_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    field_time TIME (2)
 ) ENGINE=INNODB;
 
+INSERT INTO test_table (field_varchar, field_date, field_datetime, field_time)
+VALUES ('record_zero', '1999-08-01', '2010-11-30 21:10:11.987', '30:10:15');
+
 CREATE TABLE child_table (
 	id INT AUTO_INCREMENT PRIMARY KEY,
 	parent_id INT NOT NULL,
@@ -21,6 +28,16 @@ CREATE TABLE child_table (
 	FOREIGN KEY (parent_id) REFERENCES test_table(id)
 );
 
+CREATE TABLE test_times(
+	id INT AUTO_INCREMENT PRIMARY KEY,
+	field_date DATE,
+	field_time0 TIME,
+	field_time1 TIME(1),
+	field_time6 TIME(6)
+);
+INSERT INTO test_times (field_date,   field_time0,  field_time1,   field_time6)
+VALUES 				   ('1999-08-01', '-838:59:59', '838:59:58.9', '838:59:58.999999');
+
 -- Users
 DROP USER IF EXISTS empty_password_user;
 CREATE USER empty_password_user IDENTIFIED BY '';
diff --git a/test/metadata.cpp b/test/metadata.cpp
index 8dc07c63..3584a225 100644
--- a/test/metadata.cpp
+++ b/test/metadata.cpp
@@ -20,7 +20,7 @@ namespace
 // for column_definition
 struct FieldMetadataTest : public Test
 {
-	using metadata_type = field_metadata<std::allocator<std::uint8_t>>;
+	using metadata_type = owning_field_metadata<std::allocator<std::uint8_t>>;
 
 	metadata_type make_metadata(std::vector<std::uint8_t>&& buffer) const
 	{