#include <cassert>
#include <iostream>
#include <sstream>
#include <stdexcept>

#include "clickhouse/columns/date.h"
#include "clickhouse/columns/factory.h"
#include "clickhouse/columns/ip4.h"
#include "clickhouse/columns/lowcardinality.h"
#include "clickhouse/columns/nullable.h"
#include <clickhouse/client.h>
#include <clickhouse/query.h>
#include <clickhouse/types/types.h>

#if __cplusplus > 199711L
#define register /* Deprecated in C++11. */
#endif			 /* #if __cplusplus > 199711L */

extern "C"
{

#include "postgres.h"

#include "fmgr.h"
#include "funcapi.h"
#include "internal.h"
#include "pgtime.h"

#include "access/htup_details.h"
#include "access/tupdesc.h"
#include "catalog/pg_type_d.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/date.h"
#include "utils/elog.h"
#include "utils/inet.h"
#include "utils/lsyscache.h"
#include "utils/memdebug.h"
#include "utils/palloc.h"
#include "utils/timestamp.h"
#include "utils/uuid.h"

#include "binary.hh"

	using namespace clickhouse;

#if defined(__APPLE__) /* Byte ordering on macOS */
#include <libkern/OSByteOrder.h>
#include <machine/endian.h>
#define HOST_TO_BIG_ENDIAN_64(x) OSSwapHostToBigInt64(x)
#define BIG_ENDIAN_64_TO_HOST(x) OSSwapBigToHostInt64(x)
#else
#include <endian.h>
#define HOST_TO_BIG_ENDIAN_64(x) htobe64(x)
#define BIG_ENDIAN_64_TO_HOST(x) be64toh(x)
#endif

#define THROW_UNEXPECTED_COLUMN(exp_type, col)                                                                         \
	throw std::runtime_error("unexpected column type for " + std::string(exp_type) + ": " + col->Type()->GetName())

	/* palloc which will throw exceptions */
	static void *
	exc_palloc(Size size)
	{
		/* duplicates MemoryContextAlloc to avoid increased overhead */
		void		 *ret;
		MemoryContext context = CurrentMemoryContext;

		Assert(MemoryContextIsValid(context));

		if (!AllocSizeIsValid(size))
			throw std::bad_alloc();

		context->isReset = false;

#if PG_VERSION_NUM >= 170000
		ret = context->methods->alloc(context, size, 0);
#else
		ret = context->methods->alloc(context, size);
#endif
		if (unlikely(ret == NULL))
			throw std::bad_alloc();

		VALGRIND_MEMPOOL_ALLOC(context, ret, size);

		return ret;
	}

	void *
	exc_palloc0(Size size)
	{
		/* duplicates MemoryContextAllocZero to avoid increased overhead */
		void		 *ret;
		MemoryContext context = CurrentMemoryContext;

		Assert(MemoryContextIsValid(context));

		if (!AllocSizeIsValid(size))
			throw std::bad_alloc();

		context->isReset = false;

#if PG_VERSION_NUM >= 170000
		ret = context->methods->alloc(context, size, 0);
#else
		ret = context->methods->alloc(context, size);
#endif
		if (unlikely(ret == NULL))
			throw std::bad_alloc();

		VALGRIND_MEMPOOL_ALLOC(context, ret, size);

		MemSetAligned(ret, 0, size);

		return ret;
	}

#define CLICKHOUSE_SECURE_PORT 9440

	ch_binary_connection_t *
	ch_binary_connect(ch_connection_details *details, char **error)
	{
		ClientOptions		   *options = NULL;
		ch_binary_connection_t *conn = NULL;

		try
		{
			options = new ClientOptions();
			options->SetPingBeforeQuery(true);

			if (details->host)
			{
				options->SetHost(std::string(details->host));
				if (!details->port && ch_is_cloud_host(details->host))
					options->SetPort(CLICKHOUSE_SECURE_PORT);
			}
			if (details->port)
				options->SetPort(details->port);
			if (details->dbname)
				options->SetDefaultDatabase(std::string(details->dbname));
			if (details->username)
				options->SetUser(std::string(details->username));
			if (details->password)
				options->SetPassword(std::string(details->password));
			if (options->port == CLICKHOUSE_SECURE_PORT)
				options->SetSSLOptions(ClientOptions::SSLOptions());

			/* options->SetRethrowException(false); */
			conn = new ch_binary_connection_t();

			Client *client = new Client(*options);
			conn->client = client;
			conn->options = options;
		}
		catch (const std::exception &e)
		{
			if (error)
				*error = strdup(e.what());

			if (conn != NULL)
				delete conn;

			if (options != NULL)
				delete options;

			conn = NULL;
		}
		return conn;
	}

	static void
	set_resp_error(ch_binary_response_t *resp, const char *str)
	{
		if (resp->error)
			return;

		resp->error = (char *)malloc(strlen(str) + 1);
		strcpy(resp->error, str);
	}

	/*
	 * Converts query->settings to QuerySettings.
	 */
	static QuerySettings
	ch_binary_settings(const ch_query *query)
	{
		kv_iter iter;
		auto	res = QuerySettings{};

		for (iter = new_kv_iter(query->settings); !kv_iter_done(&iter); kv_iter_next(&iter))
		{
			res.insert_or_assign(iter.name, QuerySettingsField{ iter.value, 1 });
		}

		return res;
	}

	/*
	 * Converts query->param_values to QueryParams.
	 */
	static QueryParams
	ch_binary_params(const ch_query *query)
	{
		int	 i;
		auto res = QueryParams{};

		for (i = 0; i < query->num_params; i++)
			res.insert_or_assign(psprintf("p%d", i + 1), QueryParamValue(query->param_values[i]));

		return res;
	}

	static void
	set_state_error(ch_binary_read_state_t *state, const char *str)
	{
		assert(state->error == NULL);
		state->error = (char *)malloc(strlen(str) + 1);
		strcpy(state->error, str);
	}

	ch_binary_response_t *
	ch_binary_simple_query(ch_binary_connection_t *conn, const ch_query *query, bool (*check_cancel)(void))
	{
		Client											*client = (Client *)conn->client;
		ch_binary_response_t							*resp;
		std::vector<std::vector<clickhouse::ColumnRef>> *values;

		try
		{
			resp = new ch_binary_response_t();
			values = new std::vector<std::vector<clickhouse::ColumnRef>>();
			client->Select(clickhouse::Query(query->sql)
						   .SetQuerySettings(ch_binary_settings(query))
						   .SetParams(ch_binary_params(query))
						   .OnDataCancelable(
						   [&resp, &values, &check_cancel](const Block &block)
						   {
							   if (check_cancel && check_cancel())
							   {
								   set_resp_error(resp, "query was canceled");
								   return false;
							   }

							   /* some empty block */
							   if (block.GetColumnCount() == 0)
								   return true;

							   auto vec = std::vector<clickhouse::ColumnRef>();

							   if (resp->columns_count && block.GetColumnCount() != resp->columns_count)
							   {
								   set_resp_error(resp, "columns mismatch in blocks");
								   return false;
							   }

							   resp->columns_count = block.GetColumnCount();
							   resp->blocks_count++;

							   for (size_t i = 0; i < resp->columns_count; ++i)
								   vec.push_back(block[i]);

							   values->push_back(std::move(vec));
							   return true;
						   }));

			resp->values = (void *)values;
		}
		catch (const std::exception &e)
		{
			client->ResetConnection();

			values->clear();
			set_resp_error(resp, e.what());
			delete values;
			values = NULL;
		}

		resp->success = (resp->error == NULL);
		return resp;
	}

	static Oid
	get_corr_postgres_type(const TypeRef &type)
	{
		switch (type->GetCode())
		{
			case Type::Code::Int8:
			case Type::Code::Int16:
			case Type::Code::UInt8:
				return INT2OID;
			case Type::Code::Int32:
			case Type::Code::UInt16:
				return INT4OID;
			case Type::Code::Int64:
			case Type::Code::UInt64:
			case Type::Code::UInt32:
				return INT8OID;
			case Type::Code::Float32:
				return FLOAT4OID;
			case Type::Code::Float64:
				return FLOAT8OID;
			case Type::Code::Decimal128:
			case Type::Code::Decimal64:
			case Type::Code::Decimal32:
			case Type::Code::Decimal:
				return NUMERICOID;
			case Type::Code::FixedString:
			case Type::Code::Enum8:
			case Type::Code::Enum16:
			case Type::Code::String:
				return TEXTOID;
			case Type::Code::LowCardinality:
				return get_corr_postgres_type(type->As<LowCardinalityType>()->GetNestedType());
			case Type::Code::Date:
			case Type::Code::Date32:
				return DATEOID;
			case Type::Code::DateTime:
				return TIMESTAMPTZOID;
			case Type::Code::DateTime64:
				return TIMESTAMPTZOID;
			case Type::Code::UUID:
				return UUIDOID;
			case Type::Code::Array:
			{
				Oid array_type
				= get_array_type(get_corr_postgres_type(type->As<clickhouse::ArrayType>()->GetItemType()));
				if (array_type == InvalidOid)
					throw std::runtime_error("pg_clickhouse: could not find array "
											 " type for column type "
											 + type->GetName());

				return array_type;
			}
			case Type::Code::Tuple:
				return RECORDOID;
			case Type::Code::Nullable:
				return get_corr_postgres_type(type->As<NullableType>()->GetNestedType());
			case Type::Code::IPv4:
			case Type::Code::IPv6:
				return INETOID;
			default:
				throw std::runtime_error("pg_clickhouse: unsupported column type " + type->GetName());
		}
	}

	void
	ch_binary_insert_state_free(void *c)
	{
		auto *state = (ch_binary_insert_state *)c;
		if (state->insert_block)
		{
			/* Finish the insert to set the proper ClickHouse state */
			delete (Block *)state->insert_block;
			Client *client = (Client *)state->conn->client;
			try
			{
				client->EndInsert();
			}
			catch (const std::exception &e)
			{
				client->ResetConnection();
				elog(ERROR, "pg_clickhouse: could not finish INSERT - %s", e.what());
			}
		}
	}

	void
	ch_binary_prepare_insert(void *conn, const ch_query *query, ch_binary_insert_state *state)
	{
		/* Start the INSERT. */
		Block  *block;
		Client *client = (Client *)((ch_binary_connection_t *)conn)->client;
		try
		{
			block = new Block(client->BeginInsert(std::string(query->sql) + " VALUES"));
			/* XXX https://github.com/ClickHouse/clickhouse-cpp/pull/453/
			block = new Block(client->BeginInsert(
				clickhouse::Query(std::string(query->sql)+ " VALUES").SetQuerySettings(
					ch_binary_settings(query)
				).SetParams(
					ch_binary_params(query)
				)
			));
			*/
		}
		catch (const std::exception &e)
		{
			client->ResetConnection();
			elog(ERROR, "pg_clickhouse: could not prepare insert - %s", e.what());
		}

		/* Setup the column config (or return if no columns). */
		state->len = block->GetColumnCount();
		if (state->len == 0)
		{
			delete block;
			return;
		}
		state->outdesc = CreateTemplateTupleDesc(state->len);

		/* Iterate over the list of columns returned by ClickHouse. */
		AttrNumber i = 0;
		for (Block::Iterator bi(*block); bi.IsValid(); bi.Next())
		{
			Oid			pg_type;
			const char *colname;
			try
			{
				/* Determine the Postgres column type. */
				pg_type = get_corr_postgres_type(bi.Type());
				colname = bi.Name().c_str();
			}
			catch (const std::exception &e)
			{
				elog(ERROR, "pg_clickhouse: could not prepare insert - %s", e.what());
			}
			PG_TRY();
			{
				TupleDescInitEntry(state->outdesc, ++i, colname, pg_type, -1, 0);
			}
			PG_CATCH();
			{
				/* Clean up and re-throw. */
				client->ResetConnection();
				delete block;
				PG_RE_THROW();
			}
			PG_END_TRY();
		}

		state->insert_block = (ch_insert_block_h *)block;
	}

	static void
	column_append(clickhouse::ColumnRef col, Datum val, Oid valtype, bool isnull)
	{
		bool nullable = false;

		if (col->Type()->GetCode() == Type::Code::Nullable)
			nullable = true;

		/*
		 * Prevent insertion if the column isn't Nullable. This differs from the
		 * http engine, which isn't aware of Nullable but just sends values off,
		 * in which case ClickHouse inserts default values (e.g., 0 for numbers,
		 * "" for strings, etc.).
		 *
		 * XXX Do we want to consider rejecting NULL inserts when the Postgres
		 * `NOT NULL` constraint exits? And when it doesn't exist and the column
		 * is not nullable, do we want to send the default value instead? Dropping
		 * this block should do that.
		 */
		if (isnull && !nullable)
			throw std::runtime_error("cannot append NULL to NOT NULL " + col->Type()->GetName() + " column");

		if (nullable)
		{
			auto nullable = col->AsStrict<ColumnNullable>();
			nullable->Append(isnull);
			col = nullable->Nested();
		}

		switch (valtype)
		{
			case INT2OID:
			{
				switch (col->Type()->GetCode())
				{
					case Type::Code::UInt8:
						col->AsStrict<ColumnUInt8>()->Append((uint8_t)val);
						break;
					case Type::Code::Int8:
						col->AsStrict<ColumnInt8>()->Append((int8_t)val);
						break;
					case Type::Code::Int16:
						col->AsStrict<ColumnInt16>()->Append((int16_t)val);
						break;
					default:
						THROW_UNEXPECTED_COLUMN("INT2", col);
				}
				break;
			}
			case INT4OID:
			{
				switch (col->Type()->GetCode())
				{
					case Type::Code::Int32:
						col->AsStrict<ColumnInt32>()->Append((int32_t)val);
						break;
					case Type::Code::UInt16:
						col->AsStrict<ColumnUInt16>()->Append((uint16_t)val);
						break;
					default:
						THROW_UNEXPECTED_COLUMN("INT4", col);
				}
				break;
			}
			case INT8OID:
			{
				switch (col->Type()->GetCode())
				{
					case Type::Code::Int64:
						col->AsStrict<ColumnInt64>()->Append((int64_t)val);
						break;
					case Type::Code::UInt32:
						col->AsStrict<ColumnUInt32>()->Append((uint32_t)val);
						break;
					case Type::Code::UInt64:
						col->AsStrict<ColumnUInt64>()->Append((uint64_t)val);
						break;
					default:
						THROW_UNEXPECTED_COLUMN("INT8", col);
				}
				break;
			}
			case FLOAT4OID:
			{
				switch (col->Type()->GetCode())
				{
					case Type::Code::Float32:
						col->AsStrict<ColumnFloat32>()->Append(DatumGetFloat4(val));
						break;
					default:
						THROW_UNEXPECTED_COLUMN("FLOAT4", col);
				}
				break;
			}
			case FLOAT8OID:
			{
				switch (col->Type()->GetCode())
				{
					case Type::Code::Float64:
						col->AsStrict<ColumnFloat64>()->Append(DatumGetFloat8(val));
						break;
					default:
						THROW_UNEXPECTED_COLUMN("FLOAT8", col);
				}
				break;
			}
			case NUMERICOID:
			{
				/* Convert numeric to string and let ColumnDecimal parse it. */
				switch (col->Type()->GetCode())
				{
					case Type::Code::Decimal128:
					case Type::Code::Decimal64:
					case Type::Code::Decimal32:
					case Type::Code::Decimal:
						if (isnull)
							col->AsStrict<ColumnDecimal>()->Append(Int128{});
						else
						{
							char *s = DatumGetCString(DirectFunctionCall1(numeric_out, val));
							col->AsStrict<ColumnDecimal>()->Append(std::string(s));
							pfree(s);
						}
						break;
					default:
						THROW_UNEXPECTED_COLUMN("NUMERIC", col);
				}
				break;
			}
			case TEXTOID:
			{
				/*
				 * ClickHouse allows nuls in strings, and val can be bytea, so
				 * don't rely on nul termination but copy the full length of the
				 * data.
				 */
				std::string s;
				if (!isnull)
				{
					text *string = PG_DETOAST_DATUM(val);
					s.assign(VARDATA(string), VARSIZE_ANY_EXHDR(string));
				}

				switch (col->Type()->GetCode())
				{
					case Type::Code::FixedString:
						col->AsStrict<ColumnFixedString>()->Append(s);
						break;
					case Type::Code::String:
						col->AsStrict<ColumnString>()->Append(s);
						break;
					case Type::Code::Enum8:
						if (isnull)
							col->AsStrict<ColumnEnum8>()->Append(0, false);
						else
							col->AsStrict<ColumnEnum8>()->Append(s);
						break;
					case Type::Code::Enum16:
						if (isnull)
							col->AsStrict<ColumnEnum16>()->Append(0, false);
						else
							col->AsStrict<ColumnEnum16>()->Append(s);
						break;
					case Type::Code::LowCardinality:
						/* XXX Handle LowCardinality(Nullable(x)) */
						if (col->AsStrict<ColumnLowCardinality>()->GetNestedType()->GetCode() == Type::Nullable)
							throw std::runtime_error("nested Nullable is not supported");
						col->AsStrict<ColumnLowCardinalityT<ColumnString>>()->Append(s);
						break;
					default:
						THROW_UNEXPECTED_COLUMN("TEXT", col);
				}

				break;
			}
			case DATEOID:
			{
				Timestamp t = date2timestamp_no_overflow(DatumGetDateADT(val));
				pg_time_t d = timestamptz_to_time_t(t);

				switch (col->Type()->GetCode())
				{
					case Type::Code::Date:
						col->AsStrict<ColumnDate>()->Append(d);
						break;
					case Type::Code::Date32:
						col->AsStrict<ColumnDate32>()->Append(d);
						break;
					default:
						THROW_UNEXPECTED_COLUMN("DATE", col);
				}
				break;
			}
			case TIMESTAMPOID:
			case TIMESTAMPTZOID:
			{
				switch (col->Type()->GetCode())
				{
					case Type::Code::DateTime:
					{
						pg_time_t d = timestamptz_to_time_t(DatumGetTimestamp(val));
						col->AsStrict<ColumnDateTime>()->Append(d);
						break;
					}
					case Type::Code::DateTime64:
					{
						auto	  dt64_col = col->AsStrict<ColumnDateTime64>();
						Timestamp t = DatumGetTimestamp(val);
						Int64	  dt64
						= ((1.0 * t) / USECS_PER_SEC + ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY))
						  * pow(10.0, dt64_col->GetPrecision());

						dt64_col->Append(dt64);
						break;
					}
					default:
						THROW_UNEXPECTED_COLUMN("TIMESTAMP", col);
				}
				break;
			}
			case ANYARRAYOID:
			{
				switch (col->Type()->GetCode())
				{
					case Type::Array:
					{
						auto arrcol = col->AsStrict<ColumnArray>();
						auto items
						= CreateColumnByType(arrcol->GetType().As<clickhouse::ArrayType>()->GetItemType()->GetName());
						auto arr = (ch_binary_array_t *)DatumGetPointer(val);
						for (size_t i = 0; i < arr->len; i++)
							column_append(items, arr->datums[i], arr->item_type, arr->nulls[i]);

						arrcol->AppendAsColumn(items);
						break;
					}
					default:
						THROW_UNEXPECTED_COLUMN("array", col);
				}
				break;
			}
			case UUIDOID:
			{
				auto uuid_val = DatumGetUUIDP(val)->data;
				auto ch_uuid = UUID{};
				if (!isnull)
				{
					memcpy(&ch_uuid.first, uuid_val, 8);
					memcpy(&ch_uuid.second, uuid_val + 8, 8);
					ch_uuid.first = BIG_ENDIAN_64_TO_HOST(ch_uuid.first);
					ch_uuid.second = BIG_ENDIAN_64_TO_HOST(ch_uuid.second);
				}
				col->AsStrict<ColumnUUID>()->Append(ch_uuid);
			}
			break;
			case INETOID:
			{
				if (isnull)
				{
					switch (col->Type()->GetCode())
					{
						case Type::Code::IPv4:
							col->AsStrict<ColumnIPv4>()->Append(0);
							break;
						case Type::Code::IPv6:
							col->AsStrict<ColumnIPv6>()->Append("::");
							break;
						default:
							THROW_UNEXPECTED_COLUMN("INET", col);
					}
				}
				else
				{
					char *s = DatumGetCString(DirectFunctionCall1(inet_out, val));
					switch (col->Type()->GetCode())
					{
						case Type::Code::IPv4:
							col->AsStrict<ColumnIPv4>()->Append(s);
							break;
						case Type::Code::IPv6:
							col->AsStrict<ColumnIPv6>()->Append(s);
							break;
						default:
							THROW_UNEXPECTED_COLUMN("INET", col);
					}
				}
			}
			break;
			default:
			{
				THROW_UNEXPECTED_COLUMN(std::to_string(valtype), col);
			}
		}
	}

	void
	ch_binary_column_append_data(ch_binary_insert_state *state, size_t colidx)
	{
		try
		{
			auto block = (Block *)state->insert_block;
			auto col = (*block)[colidx];

			Datum val = state->values[colidx];
			Oid	  valtype = TupleDescAttr(state->outdesc, colidx)->atttypid;
			bool  isnull = state->nulls[colidx];

			column_append(col, val, valtype, isnull);
		}
		catch (const std::exception &e)
		{
			elog(ERROR, "pg_clickhouse: could not append data to column - %s", e.what());
		}
	}

	void
	ch_binary_insert_columns(ch_binary_insert_state *state)
	{
		Client *client = (Client *)state->conn->client;
		auto	block = (Block *)state->insert_block;
		try
		{
			block->RefreshRowCount();
			client->SendInsertBlock(*block);
			block->Clear();
		}
		catch (const std::exception &e)
		{
			client->ResetConnection();
			delete block;
			elog(ERROR, "pg_clickhouse: could not insert columns - %s", e.what());
		}
	}

	void
	ch_binary_close(ch_binary_connection_t *conn)
	{
		delete (Client *)conn->client;
		delete (ClientOptions *)conn->options;
	}

	void
	ch_binary_response_free(ch_binary_response_t *resp)
	{
		if (resp->values)
		{
			auto values = (std::vector<std::vector<clickhouse::ColumnRef>> *)resp->values;
			values->clear();
			delete values;
		}

		if (resp->error)
			free(resp->error);

		delete resp;
	}

	void
	ch_binary_read_state_init(ch_binary_read_state_t *state, ch_binary_response_t *resp)
	{
		state->resp = resp;
		state->block = 0;
		state->row = 0;
		state->done = false;
		state->error = NULL;
		state->coltypes = NULL;
		state->values = NULL;
		state->nulls = NULL;

		/* it response was errored just set error in state too */
		if (resp->error)
		{
			state->done = true;
			set_state_error(state, resp->error);
			return;
		}

		try
		{
			assert(resp->values);
			auto &values = *((std::vector<std::vector<clickhouse::ColumnRef>> *)resp->values);

			if (resp->columns_count && values.size() > 0)
			{
				state->coltypes = new Oid[resp->columns_count];
				state->values = new Datum[resp->columns_count];
				state->nulls = new bool[resp->columns_count];
			}
		}
		catch (const std::exception &e)
		{
			set_state_error(state, e.what());
		}
	}

	/*
	 * This function is preparing values for `convert_datum` which is called in upper
	 * code.
	 *
	 * This function calls postgres functions, which can call `palloc` so we can end up
	 * with elog(ERROR) and longjmp to upper postgres code with leaking c++ memory.
	 *
	 * There is not an adequate (without huge overheads) solution, we just consider
	 * this state unfixable.
	 */
	static Datum
	make_datum(clickhouse::ColumnRef col, size_t row, Oid *valtype, bool *is_null)
	{
		Datum ret = (Datum)0;

	nested_col:
		auto type_code = col->Type()->GetCode();

		*valtype = InvalidOid;
		*is_null = false;

		switch (type_code)
		{
			case Type::Code::UInt8:
			{
				int16 val = col->AsStrict<ColumnUInt8>()->At(row);
				ret = (Datum)val;
				*valtype = INT2OID;
			}
			break;
			case Type::Code::UInt16:
			{
				int16 val = col->AsStrict<ColumnUInt16>()->At(row);
				ret = (Datum)val;
				*valtype = INT4OID;
			}
			break;
			case Type::Code::UInt32:
			{
				int64 val = col->AsStrict<ColumnUInt32>()->At(row);
				ret = Int64GetDatum(val);
				*valtype = INT8OID;
			}
			break;
			case Type::Code::UInt64:
			{
				uint64 val = col->AsStrict<ColumnUInt64>()->At(row);
				/* XXX Consider using, e.g., https://pgxn.org/dist/uint128. */
				if (val > LONG_MAX)
					throw std::overflow_error("value " + std::to_string(val) + " is out of range of bigint");

				ret = Int64GetDatum((int64)val);
				*valtype = INT8OID;
			}
			break;
			case Type::Code::Int8:
			{
				int16 val = col->AsStrict<ColumnInt8>()->At(row);
				ret = (Datum)val;
				*valtype = INT2OID;
			}
			break;
			case Type::Code::Int16:
			{
				int16 val = col->AsStrict<ColumnInt16>()->At(row);
				ret = (Datum)val;
				*valtype = INT2OID;
			}
			break;
			case Type::Code::Int32:
			{
				int val = col->AsStrict<ColumnInt32>()->At(row);
				ret = (Datum)val;
				*valtype = INT4OID;
			}
			break;
			case Type::Code::Int64:
			{
				int64 val = col->AsStrict<ColumnInt64>()->At(row);
				ret = Int64GetDatum(val);
				*valtype = INT8OID;
			}
			break;
			case Type::Code::Float32:
			{
				float val = col->AsStrict<ColumnFloat32>()->At(row);
				ret = Float4GetDatum(val);
				*valtype = FLOAT4OID;
			}
			break;
			case Type::Code::Float64:
			{
				double val = col->AsStrict<ColumnFloat64>()->At(row);
				ret = Float8GetDatum(val);
				*valtype = FLOAT8OID;
			}
			break;
			case Type::Code::Decimal128:
			case Type::Code::Decimal64:
			case Type::Code::Decimal32:
			case Type::Code::Decimal:
			{
				auto decCol = col->AsStrict<ColumnDecimal>();
				auto val = decCol->At(row);

				/* Convert the Int128 to a string. */
				std::stringstream ss;
				ss << val;
				std::string str = ss.str();

				/* Start a destination string. */
				std::stringstream res;
				auto			  scale = decCol->GetScale();

				/* Output a dash for negative values */
				if (val < 0)
				{
					res << '-';
					str.erase(0, 1);
				}

				if (scale == 0)
				{
					/* No decimal point, just output the entire value. */
					res << str;
				}
				else if (str.length() <= scale)
				{
					/* Append the entire value prepended with zeros after the decimal. */
					res << "0." << std::string(scale - str.length(), '0') << str;
				}
				else
				{
					/* There are digits before the decimal. */
					auto decAt = str.length() - scale;
					res << str.substr(0, decAt);

					/* Append any digits after the decimal. */
					if (decAt < str.length())
					{
						res << '.' << str.substr(decAt);
					}
				}

				ret = DirectFunctionCall3(numeric_in, CStringGetDatum(res.str().c_str()), ObjectIdGetDatum(0),
										  Int32GetDatum(-1));
				*valtype = NUMERICOID;
			}
			break;
			case Type::Code::FixedString:
			{
				auto s = std::string(col->AsStrict<ColumnFixedString>()->At(row));
				/* ClickHouse allows nuls in strings, so copy by full length. */
				ret = PointerGetDatum(cstring_to_text_with_len(s.data(), s.size()));
				*valtype = TEXTOID;
			}
			break;
			case Type::Code::String:
			{
				auto s = std::string(col->AsStrict<ColumnString>()->At(row));
				/* ClickHouse allows nuls in strings, so copy by full length. */
				ret = PointerGetDatum(cstring_to_text_with_len(s.data(), s.size()));
				*valtype = TEXTOID;
			}
			break;
			case Type::Code::Enum8:
			{
				auto s = std::string(col->AsStrict<ColumnEnum8>()->NameAt(row));
				ret = CStringGetTextDatum(s.c_str());
				*valtype = TEXTOID;
			}
			break;
			case Type::Code::Enum16:
			{
				auto s = std::string(col->AsStrict<ColumnEnum16>()->NameAt(row));
				ret = CStringGetTextDatum(s.c_str());
				*valtype = TEXTOID;
			}
			break;
			case Type::Code::Date:
			{
				auto val = static_cast<pg_time_t>(col->AsStrict<ColumnDate>()->At(row));
				*valtype = DATEOID;
				ret = DirectFunctionCall1(timestamp_date, time_t_to_timestamptz(val));
			}
			break;
			case Type::Code::Date32:
			{
				auto val = static_cast<pg_time_t>(col->AsStrict<ColumnDate32>()->At(row));
				*valtype = DATEOID;
				ret = DirectFunctionCall1(timestamp_date, time_t_to_timestamptz(val));
			}
			break;
			case Type::Code::DateTime:
			{
				auto val = static_cast<pg_time_t>(col->AsStrict<ColumnDateTime>()->At(row));
				*valtype = TIMESTAMPTZOID;
				ret = TimestampTzGetDatum(time_t_to_timestamptz(val));
			}
			break;
			case Type::Code::DateTime64:
			{
				auto  dt_col = col->AsStrict<ColumnDateTime64>();
				auto  val = dt_col->At(row);
				int64 power = pow(10, dt_col->GetPrecision());
				*valtype = TIMESTAMPTZOID;
				ret = TimestampTzGetDatum(time_t_to_timestamptz(val / power)) + (val % power) * (USECS_PER_SEC / power);
			}
			break;
			case Type::Code::UUID:
			{
				/* we form char[16] from two uint64 numbers, and they should
				 * be big endian */
				auto	   val = col->AsStrict<ColumnUUID>()->At(row);
				pg_uuid_t *uuid_val = (pg_uuid_t *)exc_palloc(sizeof(pg_uuid_t));

				val.first = HOST_TO_BIG_ENDIAN_64(val.first);
				val.second = HOST_TO_BIG_ENDIAN_64(val.second);
				memcpy(uuid_val->data, &val.first, 8);
				memcpy(uuid_val->data + 8, &val.second, 8);

				ret = UUIDPGetDatum(uuid_val);
				*valtype = UUIDOID;
			}
			break;
			case Type::Code::Nullable:
			{
				auto nullable = col->AsStrict<ColumnNullable>();
				if (nullable->IsNull(row))
				{
					*is_null = true;
				}
				else
				{
					col = nullable->Nested();
					goto nested_col;
				}
			}
			break;
			case Type::Code::Array:
			{
				auto   arr = col->AsStrict<ColumnArray>()->GetAsColumn(row);
				size_t len = arr->Size();
				auto   slot = (ch_binary_array_t *)exc_palloc(sizeof(ch_binary_array_t));

				Oid item_type = get_corr_postgres_type(arr->Type());
				Oid array_type = get_array_type(item_type);

				if (array_type == InvalidOid)
					throw std::runtime_error(std::string("pg_clickhouse: could not") + " find array type for "
											 + std::to_string(item_type));

				slot->len = len;
				slot->array_type = array_type;
				slot->item_type = item_type;

				if (len > 0)
				{
					slot->datums = (Datum *)exc_palloc0(sizeof(Datum) * len);
					slot->nulls = (bool *)exc_palloc0(sizeof(bool) * len);

					for (size_t i = 0; i < len; ++i)
						slot->datums[i] = make_datum(arr, i, &slot->item_type, &slot->nulls[i]);
				}

				/* this one will need additional work, since we just return raw slot */
				ret = PointerGetDatum(slot);
				*valtype = ANYARRAYOID;
			}
			break;
			case Type::Code::Tuple:
			{
				auto tuple = col->AsStrict<ColumnTuple>();
				auto len = tuple->TupleSize();

				if (len == 0)
					throw std::runtime_error("pg_clickhouse: returned tuple is empty");

				auto slot = (ch_binary_tuple_t *)exc_palloc(sizeof(ch_binary_tuple_t));

				slot->datums = (Datum *)exc_palloc(sizeof(Datum) * len);
				slot->nulls = (bool *)exc_palloc0(sizeof(bool) * len);
				slot->types = (Oid *)exc_palloc0(sizeof(Oid) * len);
				slot->len = len;

				for (size_t i = 0; i < len; ++i)
					slot->datums[i] = make_datum((*tuple)[i], row, &slot->types[i], &slot->nulls[i]);

				/* this one will need additional work, since we just return raw slot */
				ret = PointerGetDatum(slot);
				*valtype = RECORDOID;
			}
			break;
			case Type::Code::LowCardinality:
			{
				auto item = col->AsStrict<ColumnLowCardinality>()->GetItem(row);
				auto data = item.AsBinaryData();
				ret = PointerGetDatum(cstring_to_text_with_len(data.data(), data.size()));
				*valtype = TEXTOID;
			}
			break;
			case Type::Code::IPv4:
			{
				auto item = col->AsStrict<ColumnIPv4>()->AsString(row);
				ret = DirectFunctionCall1(inet_in, CStringGetDatum(item.c_str()));
				*valtype = INETOID;
			}
			break;
			case Type::Code::IPv6:
			{
				auto item = col->AsStrict<ColumnIPv6>()->AsString(row);
				ret = DirectFunctionCall1(inet_in, CStringGetDatum(item.c_str()));
				*valtype = INETOID;
			}
			break;
			default:
				throw std::runtime_error("unsupported type " + std::string(Type::TypeName(type_code))
										 + " in binary protocol");
		}

		return ret;
	}

	bool
	ch_binary_read_row(ch_binary_read_state_t *state)
	{
		/* coltypes is NULL means there are no blocks */
		bool res = false;

		if (state->done || state->coltypes == NULL || state->error)
			return false;

		assert(state->resp->values);
		auto &values = *((std::vector<std::vector<clickhouse::ColumnRef>> *)state->resp->values);
		try
		{
		again:
			assert(state->block < state->resp->blocks_count);
			auto  &block = values[state->block];
			size_t row_count = block[0]->Size();

			if (row_count == 0)
				goto next_row;

			for (size_t i = 0; i < state->resp->columns_count; i++)
			{
				/* fill value and null arrays */
				state->values[i] = make_datum(block[i], state->row, &state->coltypes[i], &state->nulls[i]);
			}
			res = true;

		next_row:
			state->row++;
			if (state->row >= row_count)
			{
				state->row = 0;
				state->block++;
				if (state->block >= state->resp->blocks_count)
					state->done = true;
				else if (row_count == 0)
					goto again;
			}
		}
		catch (const std::exception &e)
		{
			set_state_error(state, e.what());
		}

		return res;
	}

	void
	ch_binary_read_state_free(ch_binary_read_state_t *state)
	{
		if (state->coltypes)
		{
			delete[] state->coltypes;
			delete[] state->values;
			delete[] state->nulls;
		}

		if (state->error)
			free(state->error);
	}
}
