#include "stdafx.h"
#include "PostgreSQL.h"
#include "Exception.h"
#include "Core/Convert.h"
#include "SQL/StrUtils.h"
#include "Oid.h"

namespace sql {

	static void addQuoted(StrBuf *to, const wchar *str) {
		to->addRaw('\'');
		for (const wchar *at = str; *at; at++) {
			if (*at == '\'' || *at == '\\')
				to->addRaw('\\');
			to->addRaw(*at);
		}
		to->addRaw('\'');
	}
	static void addQuoted(StrBuf *to, Str *str) {
		addQuoted(to, str->c_str());
	}

	static void swallowNotices(void *arg, const PGresult *res) {
		(void)arg;
		(void)res;
		// We don't do anything here. We can probably get the "error message" from res if we want to
		// store it somewhere.
	}

	PostgreSQL::PostgreSQL(Host host, MAYBE(Str *) user, MAYBE(Str *) password, Str *database) {
		try {
			driver = createPGDriver(engine());

			StrBuf *info = new (this) StrBuf();
			*info << S("client_encoding=utf8");
			*info << S(" dbname=");
			addQuoted(info, database);

			if (user) {
				*info << S(" user=");
				addQuoted(info, user);
			}

			if (password) {
				*info << S(" password=");
				addQuoted(info, password);
			}

			if (Address *addr = host.isSocket()) {
				*info << S(" hostaddr=");
				addQuoted(info, addr->withPort(0)->toS());
				*info << S(" port=") << addr->port();
			} else if (Str *local = host.isLocal()) {
				*info << S(" host=");
				if (!local->startsWith(S("/")))
					local = *new (this) Str(S("/")) + local;
				addQuoted(info, local);
			}

			const char *utf8Info = info->utf8_str();
			connection = (*driver->PQconnectdb)(utf8Info);

			if ((*driver->PQstatus)(connection) != CONNECTION_OK) {
				throwError();
			}

			// Make a cancellation connection. Note that we support two different interfaces to have
			// wide compatibility across versions.
			if (driver->PQcancelCreate && driver->PQcancelFinish && driver->PQcancelBlocking) {
				// New-style:
				newCancel = (*driver->PQcancelCreate)(connection);
			} else if (driver->PQgetCancel && driver->PQfreeCancel && driver->PQcancel) {
				// Note: This interface is supposedly not thread-safe. Perhaps we should warn?
				oldCancel = (*driver->PQgetCancel)(connection);
			}

			// Set the notice receiver so that we can stop warnings from being printed to stderr. We
			// might want to handle them properly at some time.
			if (driver->PQsetNoticeReceiver) {
				(*driver->PQsetNoticeReceiver)(connection, &swallowNotices, null);
			}

		} catch (...) {
			if (connection && driver)
				(*driver->PQfinish)(connection);
			if (driver)
				destroyPGDriver(driver);
			connection = null;
			driver = null;
			throw;
		}
	}

	PostgreSQL::~PostgreSQL() {
		close();
	}

	void PostgreSQL::close() {
		DBConnection::close();
		if (driver) {
			if (newCancel)
				(*driver->PQcancelFinish)(newCancel);
			if (oldCancel)
				(*driver->PQfreeCancel)(oldCancel);

			if (connection)
				(*driver->PQfinish)(connection);
			connection = null;

			destroyPGDriver(driver);
		}
		driver = null;
	}

	void PostgreSQL::throwError() {
		if (!driver || !connection)
			return;

		const char *error = (*driver->PQerrorMessage)(connection);

		// Note: All errors end in a newline by convention.
		if (error && *error) {
			Str *message = new (this) Str(toWChar(engine(), error));
			throw new (this) SQLError(trimWhitespace(message));
		}
	}

	Statement *PostgreSQL::prepare(Str *query) {
		return new (this) Stmt(this, query);
	}

	features::DBFeatures PostgreSQL::features() const {
		return features::insertReturning
			| features::updateReturning
			| features::deleteReturning;
	}

	const char *PostgreSQL::nextName() {
		StrBuf *out = new (this) StrBuf();
		// Note: Name can not be uppercase, it seems like names are "normalized" to lowercase when
		// using them through the query API, but not through the C-api...
		*out << S("ps") << width(8) << fill('0') << nextNameId++;
		return out->utf8_str();
	}

	Array<Str *> *PostgreSQL::tables() {
		const wchar *query = S("SELECT table_name FROM information_schema.tables ")
			S("WHERE table_type = 'BASE TABLE' AND table_schema = 'public'");
		Statement *q = prepare(new (this) Str(query));

		Statement::Result r = q->execute();

		Array<Str *> *result = new (this) Array<Str *>();
		for (Maybe<Row> row = r.next(); row.any(); row = r.next()) {
			result->push(row.value().getStr(0));
		}

		q->finalize();
		return result;
	}

	static Bool isDefaultSequence(Str *value) {
		if (!value->startsWith(S("nextval('")))
			return false;
		if (!value->endsWith(S("::regclass)")))
			return false;
		return true;
	}

	static QueryType parseType(Str *typeColumn, Maybe<Int> sizeColumn) {
		if (compareNoCase(typeColumn, S("varchar")) || compareNoCase(typeColumn, S("character varying"))) {
			if (sizeColumn.any())
				return QueryType::text().sized(Nat(sizeColumn.value()));
			else
				return QueryType::integer();
		} else if (compareNoCase(typeColumn, S("text"))) {
			return QueryType::text();
		} else if (compareNoCase(typeColumn, S("smallint"))) {
			return QueryType::integer().sized(2);
		} else if (compareNoCase(typeColumn, S("integer"))) {
			return QueryType::integer().sized(4);
		} else if (compareNoCase(typeColumn, S("bigint"))) {
			return QueryType::integer().sized(8);
		} else if (compareNoCase(typeColumn, S("real"))) {
			return QueryType::real().sized(4);
		} else if (compareNoCase(typeColumn, S("double precision"))) {
			return QueryType::real().sized(8);
		} else {
			return QueryType();
		}
	}

	MAYBE(Schema *) PostgreSQL::schema(Str *table) {
		QueryStrBuilder *colB = new (this) QueryStrBuilder();
		colB->put(
			S("SELECT cols.column_name, cols.data_type, cols.is_nullable, cols.column_default, ")
			S("cols.is_identity, cols.identity_generation, cols.character_maximum_length, ")
			S("(SELECT tc.constraint_type FROM information_schema.table_constraints AS tc ")
			S("JOIN information_schema.constraint_column_usage AS ccu ON tc.constraint_name = ccu.constraint_name WHERE ccu.table_name = cols.table_name AND ccu.column_name = cols.column_name AND tc.table_schema = cols.table_schema ")
			S("ORDER BY CASE tc.constraint_type WHEN 'PRIMARY KEY' THEN 1 WHEN 'UNIQUE' THEN 2 WHEN 'FOREIGN KEY' THEN 3 ELSE 4 END LIMIT 1) AS constraint_type FROM information_schema.columns AS cols ")
			S("WHERE table_name = "));
		colB->quoted(table);
		colB->put(S(" ORDER BY cols.ordinal_position"));

		// Columns are:
		enum {
			colName = 0,
			dataType = 1,
			isNullable = 2,
			columnDefault = 3,
			isIdentity = 4,
			identityGeneration = 5,
			maximumCharLength = 6,
			constraintType = 7
		};

		Statement *colQuery = prepare(colB->build());
		Array<Schema::Column *> *columns = new (this) Array<Schema::Column *>();
		Statement::Result colResult = colQuery->execute();

		for (Maybe<Row> row = colResult.next(); row.any(); row = colResult.next()) {
			Row &v = row.value();

			Maybe<Int> typeSize;
			if (!v.isNull(maximumCharLength))
				typeSize = Maybe<Int>(v.getInt(maximumCharLength));
			QueryType type = parseType(v.getStr(dataType), typeSize);

			Schema::Column *col = new (this) Schema::Column(v.getStr(colName), type);

			// Do we know the type?
			if (col->type.empty())
				col->unknown = v.getStr(dataType);

			// Nullable?
			if (compareNoCase(v.getStr(isNullable), S("NO")))
				col->attributes |= Schema::notNull;

			// What kind of key?
			if (!v.isNull(constraintType)) {
				Str *key = v.getStr(constraintType);
				if (compareNoCase(key, S("PRIMARY KEY")))
					col->attributes |= Schema::primaryKey;
				else if (compareNoCase(key, S("UNIQUE")))
					col->attributes |= Schema::unique;
			}

			// Identity/autoincrement?
			// We have two cases we want to catch:
			// 1: when using the SERIAL type - this is simply an integer with a default of the form nextval(...)
			// 2: when using GENERATED BY DEFAULT AS IDENTITY, we will get 'is_identity' YES

			Str *defaultVal = null;
			if (!v.isNull(columnDefault))
				defaultVal = v.getStr(columnDefault);

			if (compareNoCase(v.getStr(isIdentity), S("YES")) &&
				compareNoCase(v.getStr(identityGeneration), S("BY DEFAULT"))) {
				col->attributes |= Schema::autoIncrement;
			} else if (defaultVal && isDefaultSequence(defaultVal)) {
				col->attributes |= Schema::autoIncrement;
				defaultVal = null; // Don't use it as a default value as well!
			}

			// Default value?
			if (defaultVal && defaultVal->any()) {
				// Remove typecast if it exists.
				Str::Iter cast = defaultVal->findLast(new (this) Str(S("::")));
				col->defaultValue = defaultVal->cut(defaultVal->begin(), cast);
			}

			columns->push(col);
		}

		colQuery->finalize();

		// No columns -> return null.
		if (columns->empty())
			return null;


		// Find indexes!
		QueryStrBuilder *indexB = new (this) QueryStrBuilder();
		indexB->put(
			S("SELECT i.relname AS index_name, ")
			S("array_to_string(array_agg(a.attname ORDER BY a.attnum), '\"') AS index_columns ") // using " as delimiter as we can't encode that character in table names anyway
			S("FROM pg_class AS t ")
			S("JOIN pg_index AS ix ON ix.indrelid = t.oid ")
			S("JOIN pg_class AS i ON i.oid = ix.indexrelid ")
			S("JOIN pg_attribute AS a ON a.attrelid = t.oid AND a.attnum = ANY(ix.indkey) ")
			S("WHERE t.relname = "));
		indexB->quoted(table);
		indexB->put(
			S(" AND t.relkind = 'r' ") // regular table
			S(" AND NOT ix.indisprimary AND NOT ix.indisunique") // exclude primary and unique indices
			S(" GROUP BY i.relname"));

		// indexB->put(
		// 	S("SELECT i.relname AS index_name, ")
		// 	S("CASE WHEN ix.indisprimary THEN 'PRIMARY KEY' WHEN ix.indisunique THEN 'UNIQUE' ELSE 'INDEX' END AS index_type, ")
		// 	S("array_to_string(array_agg(a.attname ORDER BY a.attnum), '\"') AS index_columns ") // using " as delimiter as we can't encode that character in table names anyway
		// 	S("FROM pg_class t ")
		// 	S("JOIN pg_index ix ON ix.indrelid = t.oid ")
		// 	S("JOIN pg_class i ON i.oid = ix.indexrelid ")
		// 	S("JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = ANY(ix.indkey) ")
		// 	S("WHERE t.relname = "));
		// indexB->quoted(table);
		// indexB->put(
		// 	S(" AND t.relkind = 'r' ") // regular table
		// 	S("GROUP BY i.relname, ix.indisprimary, ix.indisunique ")
		// 	S("ORDER BY i.relname"));

		Statement *indexQuery = prepare(indexB->build());
		Array<Schema::Index *> *indices = new (this) Array<Schema::Index *>();
		Statement::Result indexResult = indexQuery->execute();

		for (Maybe<Row> row = indexResult.next(); row.any(); row = indexResult.next()) {
			Row &v = row.value();
			Str *name = v.getStr(0);
			Str *columns = v.getStr(1);

			// Column names.
			Array<Str *> *cols = new (this) Array<Str *>();
			Str::Iter at = columns->begin();
			do {
				Str::Iter next = columns->find('"', at);
				cols->push(columns->cut(at, next));
				at = next;
				if (at != columns->end())
					++at;
			} while (at != columns->end());

			indices->push(new (this) Schema::Index(name, cols));
		}

		indexQuery->finalize();

		return new (this) Schema(table, columns, indices);
	}

	void PostgreSQL::migrate(Migration *m) {
		Transaction transaction(this);

		// Remove indices.
		for (Nat i = 0; i < m->indexRemove->count(); i++) {
			query(TO_S(this, S("DROP INDEX \"") << m->indexRemove->at(i)->name << S("\"")));
		}

		// Remove tables.
		for (Nat i = 0; i < m->tableRemove->count(); i++) {
			query(TO_S(this, S("DROP TABLE \"") << m->tableRemove->at(i) << S("\"")));
		}

		// Migrate tables.
		for (Nat i = 0; i < m->tableMigrate->count(); i++) {
			migrateTable(m->tableMigrate->at(i));
		}

		// Add tables.
		for (Nat i = 0; i < m->tableAdd->count(); i++) {
			Schema *table = m->tableAdd->at(i);

			QueryStrBuilder *b = new (this) QueryStrBuilder();
			table->toSQL(b);
			query(b->build());

			// Don't forget the associated indices!
			Array<Schema::Index *> *indices = table->indices();
			for (Nat j = 0; j < indices->count(); j++) {
				QueryStrBuilder *b = new (this) QueryStrBuilder();
				indices->at(j)->toSQL(b, table->name());
				query(b->build());
			}
		}

		// Add indices.
		for (Nat i = 0; i < m->indexAdd->count(); i++) {
			QueryStrBuilder *b = new (this) QueryStrBuilder();
			m->indexAdd->at(i)->toSQL(b);
			query(b->build());
		}

		transaction.commit();
	}

	static void addComma(QueryStrBuilder *to, Bool &first) {
		if (!first)
			to->put(S(", "));
		first = false;
	}

	static void alterColumn(QueryStrBuilder *to, Bool &first, Str *column, const wchar *query) {
		addComma(to, first);
		to->put(S("ALTER COLUMN "));
		to->name(column);
		to->put(S(" "));
		to->put(query);
	}

	void PostgreSQL::migrateTable(Migration::Table *m) {
		// Try to create a large query that does everything except the primary key constraints.
		// Hopefully the database will be able to do the changes as a unit in that case.
		QueryStrBuilder *b = new (this) QueryStrBuilder();
		b->put(S("ALTER TABLE "));
		b->name(m->table);
		b->put(S(" "));

		Bool firstPart = true;

		// Remove columns.
		for (Nat i = 0; i < m->colRemove->count(); i++) {
			addComma(b, firstPart);
			b->put(S("DROP COLUMN "));
			b->name(m->colRemove->at(i));
		}

		// Update columns.
		for (Nat i = 0; i < m->colMigrate->count(); i++) {
			Migration::ColAttrs *col = m->colMigrate->at(i);

			Schema::Attributes toAdd = col->desiredAttributes & ~col->currentAttributes;
			Schema::Attributes toRemove = col->currentAttributes & ~col->desiredAttributes;

			if (toRemove & Schema::notNull) {
				alterColumn(b, firstPart, col->name, S("DROP NOT NULL"));
			}
			if (toRemove & Schema::unique) {
				// We need to find the name of the constraint...
				StrBuf *query = new (this) StrBuf();
				*query << S("SELECT tc.constraint_name FROM information_schema.table_constraints AS tc ")
					   << S("JOIN information_schema.constraint_column_usage AS ccu ")
					   << S("ON tc.table_name = ccu.table_name AND tc.constraint_name = ccu.constraint_name ")
					   << S("WHERE tc.constraint_schema = 'public' AND tc.table_name = ")
					   << toSQLStrLiteral(m->table) << S(" AND ccu.column_name = ")
					   << toSQLStrLiteral(col->name) << S(" AND tc.constraint_type = 'UNIQUE'");
				Str *name = queryStr(query->toS());
				if (name) {
					addComma(b, firstPart);
					b->put(S("DROP CONSTRAINT "));
					b->name(name);
				}
			}
			if (toRemove & Schema::autoIncrement) {
				alterColumn(b, firstPart, col->name, S("DROP IDENTITY"));
			}

			if (toAdd & Schema::notNull) {
				alterColumn(b, firstPart, col->name, S("SET NOT NULL"));
			}
			if (toAdd & Schema::unique) {
				addComma(b, firstPart);
				b->put(S("ADD UNIQUE ("));
				b->name(col->name);
				b->put(S(")"));
			}
			if (toAdd & Schema::autoIncrement) {
				alterColumn(b, firstPart, col->name, S("ADD GENERATED BY DEFAULT AS IDENTITY"));
			}

			if (col->desiredDefault) {
				alterColumn(b, firstPart, col->name, S("SET DEFAULT "));
				b->put(col->desiredDefault);
			} else if (col->currentDefault) {
				alterColumn(b, firstPart, col->name, S("DROP DEFAULT"));
			}
		}

		// Add columns.
		for (Nat i = 0; i < m->colAdd->count(); i++) {
			Schema::Column *col = m->colAdd->at(i);
			addComma(b, firstPart);
			b->put(S("ADD COLUMN "));
			col->toSQL(b);
		}

		if (!firstPart)
			query(b->build());

		// Update primary keys. We do this as a separate step to avoid potential clashes with the other things.
		if (m->updatePrimaryKeys) {
			// Note that the ability to do multiple things in one ALTER TABLE query is an extension.
			QueryStrBuilder *b = new (this) QueryStrBuilder();
			b->put(S("ALTER TABLE "));
			b->name(m->table);

			Bool firstPart = true;

			if (m->dropPrimaryKeys) {
				// Figure out the primary key:
				StrBuf *pkQuery = new (this) StrBuf();
				*pkQuery << S("SELECT constraint_name FROM information_schema.table_constraints WHERE ")
						 << S("table_schema = 'public' AND table_name = ") << toSQLStrLiteral(m->table)
						 << S("AND constraint_type = 'PRIMARY KEY'");
				Str *pkName = queryStr(pkQuery->toS());
				if (pkName) {
					b->put(S(" DROP CONSTRAINT "));
					b->name(pkName);
					firstPart = false;
				}
			}

			if (m->primaryKeys->any()) {
				if (!firstPart)
					b->put(S(","));
				firstPart = false;

				b->put(S(" ADD PRIMARY KEY ("));
				b->name(m->primaryKeys->at(0));
				for (Nat i = 1; i < m->primaryKeys->count(); i++) {
					b->put(S(", "));
					b->name(m->primaryKeys->at(i));
				}
				b->put(S(")"));
			}

			query(b->build());
		}
	}

	QueryStr::Visitor *PostgreSQL::visitor() const {
		return new (this) Visitor();
	}

	void PostgreSQL::query(const char *query) {
		clearFetch();
		checkAndClear((*driver->PQexec)(connection, query));
	}

	void PostgreSQL::query(Str *q) {
		query(q->utf8_str());
	}

	void PostgreSQL::query(QueryStr *q) {
		query(q->generate(visitor())->utf8_str());
	}

	Str *PostgreSQL::queryStr(Str *q) {
		const char *query = q->utf8_str();

		clearFetch();
		PGresult *result = (*driver->PQexec)(connection, query);

		if ((*driver->PQresultStatus)(result) != PGRES_TUPLES_OK) {
			checkAndClear(result);
			throw new (this) InternalError(S("The query passed to 'queryStr' did not return any result!"));
		}

		if ((*driver->PQntuples)(result) < 1) {
			checkAndClear(result);
			return null;
		}

		char *value = (*driver->PQgetvalue)(result, 0, 0);
		// Note: Not entirely exception safe here...
		Str *resultStr = new (this) Str(toWChar(engine(), value));
		checkAndClear(result);
		return resultStr;
	}

	void PostgreSQL::beginTransaction() {
		query("BEGIN");
	}

	void PostgreSQL::endTransaction(Transaction::End end) {
		if (end == Transaction::endCommit) {
			query("COMMIT");
		} else if (end == Transaction::endRollback) {
			query("ROLLBACK");
		}
	}

	void PostgreSQL::finalizeStmt(SequentialStatement *stmt) {
		reinterpret_cast<Stmt *>(stmt)->free();
	}

	Int PostgreSQL::getLastRowId() {
		if (!lastRowIdQuery)
			lastRowIdQuery = new (this) Stmt(this, new (this) Str(S("SELECT lastval();")));

		Statement::Result res = lastRowIdQuery->execute();
		Maybe<Row> row = res.next();
		if (row.any()) {
			return row.value().getInt(0);
		} else {
			return -1;
		}
	}

	void PostgreSQL::checkAndClear(PGresult *result) {
		if (!result)
			return;

		const char *error = (*driver->PQresultErrorMessage)(result);
		if (error && *error) {
			Str *message = new (this) Str(toWChar(engine(), error));
			(*driver->PQclear)(result);
			throw new (this) SQLError(message);
		} else {
			(*driver->PQclear)(result);
		}
	}


	/**
	 * Statement.
	 */

	PostgreSQL::Stmt::Stmt(PostgreSQL *owner, Str *query)
		: SequentialStatement(owner), lastId(-1), name(owner->nextName()) {

		const char *utf8Query = query->utf8_str();

		owner->clearFetch();

		PGresult *result = (*driver().PQprepare)(owner->connection, name, utf8Query, 0, null);
		checkAndClear(result);

		result = (*driver().PQdescribePrepared)(owner->connection, name);

		try {
			paramCount = (*driver().PQnparams)(result);
			paramTypes = new Oid[paramCount];
			paramVals = new char *[paramCount];
			paramLengths = new int[paramCount];
			paramFormats = new int[paramCount];

			for (Nat i = 0; i < paramCount; i++) {
				paramTypes[i] = (*driver().PQparamtype)(result, i);
				paramVals[i] = null;
				paramLengths[i] = 0;
				paramFormats[i] = 1;
			}

			fieldCount = (*driver().PQnfields)(result);

			(*driver().PQclear)(result);
		} catch (...) {
			(*driver().PQclear)(result);
			throw;
		}
	}

	void PostgreSQL::Stmt::free() {
		for (Nat i = 0; i < paramCount; i++) {
			if (paramVals[i])
				delete [] paramVals[i];
			paramVals[i] = null;
		}

		delete [] paramTypes;
		delete [] paramVals;
		delete [] paramLengths;
		delete [] paramFormats;

		const PGDriver &d = driver();
		if (d.PQclosePrepared) {
			checkAndClear((*d.PQclosePrepared)(pgConnection(), name));
		} else {
			std::string command = "DEALLOCATE PREPARE ";
			command += name;
			checkAndClear((*d.PQexec)(pgConnection(), command.c_str()));
		}
	}

	template <class T>
	void writeType(char *bytes, T val) {
		memcpy(bytes, &val, sizeof(T));
#ifdef LITTLE_ENDIAN
		for (size_t i = 0; i < sizeof(T) / 2; i++) {
			std::swap(bytes[0], bytes[sizeof(T) - i - 1]);
		}
#endif
	}

	template <class T>
	T readType(const char *bytes) {
		T out;
#ifdef LITTLE_ENDIAN
		char copy[sizeof(T)];
		for (size_t i = 0; i < sizeof(T); i++)
			copy[sizeof(T) - i - 1] = bytes[i];
		memcpy(&out, copy, sizeof(T));
#else
		memcpy(&out, bytes, sizeof(T));
#endif
		return out;
	}

	void PostgreSQL::Stmt::bind(Nat pos, Str *str) {
		if (pos >= paramCount)
			return;

		switch (paramTypes[pos]) {
		case TEXT_OID:
		case VARCHAR_OID:
		case NAME_OID: {
			if (paramVals[pos])
				delete [] paramVals[pos];

			size_t utf8Count = convert(str->c_str(), null, 0);
			paramVals[pos] = new char[utf8Count];
			convert(str->c_str(), paramVals[pos], utf8Count);
			paramLengths[pos] = int(utf8Count - 1);

			break;
		}
		default:
			// Nothing to do.
			break;
		}
	}

	void PostgreSQL::Stmt::bind(Nat pos, Bool b) {
		if (pos >= paramCount)
			return;
		if (paramTypes[pos] != BOOL_OID)
			return;

		ensureLength(pos, 1);
		paramVals[pos][0] = b ? 1 : 0;
	}

	void PostgreSQL::Stmt::bind(Nat pos, Int i) {
		bind(pos, Long(i));
	}

	void PostgreSQL::Stmt::bind(Nat pos, Long l) {
		if (pos >= paramCount)
			return;

		switch (paramTypes[pos]) {
		case INT2_OID:
			ensureLength(pos, 2);
			writeType<short>(paramVals[pos], short(l));
			break;
		case INT4_OID:
			ensureLength(pos, 4);
			writeType<Int>(paramVals[pos], Int(l));
			break;
		case INT8_OID:
			ensureLength(pos, 8);
			writeType<Long>(paramVals[pos], Long(l));
			break;
		default:
			// Fall back to string representation to see if that works:
			bind(pos, TO_S(this, l));
			break;
		}
	}

	void PostgreSQL::Stmt::bind(Nat pos, Float f) {
		bind(pos, Double(f));
	}

	void PostgreSQL::Stmt::bind(Nat pos, Double f) {
		if (pos >= paramCount)
			return;

		switch (paramTypes[pos]) {
		case FLOAT4_OID:
			ensureLength(pos, 4);
			writeType<Float>(paramVals[pos], Float(f));
			break;
		case FLOAT8_OID:
			ensureLength(pos, 8);
			writeType<Double>(paramVals[pos], f);
			break;
		default:
			// Fall back to string representation to see if that works:
			bind(pos, TO_S(this, f));
			break;
		}
	}

	void PostgreSQL::Stmt::bindNull(Nat pos) {
		if (pos >= paramCount)
			return;

		if (paramVals[pos])
			delete [] paramVals[pos];
		paramVals[pos] = null;
		paramLengths[pos] = 0;
	}

	void PostgreSQL::Stmt::ensureLength(Nat pos, Nat length) {
		if (paramVals[pos] && Nat(paramLengths[pos]) != length) {
			delete [] paramVals[pos];
			paramVals[pos] = new char[length];
			paramLengths[pos] = int(length);
		}
	}

	Bool PostgreSQL::Stmt::executeSeq() {
		streamingData = false;

		int ok = (*driver().PQsendQueryPrepared)(pgConnection(), name, int(paramCount),
												paramVals, paramLengths, paramFormats, 1);
		if (!ok)
			owner()->throwError();

		ok = (*driver().PQsetSingleRowMode)(pgConnection());
		if (!ok)
			owner()->throwError();

		if (fieldCount == 0) {
			// We won't get a result, just wait for the result now and report that there will not be
			// any more results.
			PGresult *result = (*driver().PQgetResult)(pgConnection());
			if (!result)
				owner()->throwError();

			char *modified = (*driver().PQcmdTuples)(result);
			if (modified && *modified) {
				lastChanges = Nat(atol(modified));
			}

			bool updateRowId = strncmp((*driver().PQcmdStatus)(result), "INSERT", 6) == 0;
			checkAndClear(result);

			// Call disposeresult regardless since we don't produce any result.
			disposeResult();

			// Get the last row ID.
			if (updateRowId) {
				lastId = owner()->getLastRowId();
			}

			return false;
		} else {
			// We have results, tell the world
			streamingData = true;
			return true;
		}
	}

	Maybe<Row> PostgreSQL::Stmt::nextRowSeq() {
		PGresult *result = (*driver().PQgetResult)(pgConnection());
		if (!result) {
			// We are done.
			streamingData = false;
			return Maybe<Row>();
		}

		try {
			ExecStatusType status = (*driver().PQresultStatus)(result);

			if (status == PGRES_COMMAND_OK || status == PGRES_TUPLES_OK) {
				// Note: We should technically call PQgetResult more it seems.
				streamingData = false;
				checkAndClear(result);
				return Maybe<Row>();
			}

			Row::Builder builder = Row::builder(engine(), fieldCount);

			for (Nat i = 0; i < fieldCount; i++) {
				Oid type = (*driver().PQftype)(result, int(i));
				const char *value = (*driver().PQgetvalue)(result, 0, int(i));
				int isNull = (*driver().PQgetisnull)(result, 0, int(i));
				int length = (*driver().PQgetlength)(result, 0, int(i));

				if (isNull) {
					builder.pushNull();
					continue;
				}

				switch (type) {
				case TEXT_OID:
				case VARCHAR_OID:
				case NAME_OID:
					builder.push(new (this) Str(toWChar(engine(), value, size_t(length))));
					break;
				case BOOL_OID:
					builder.push(Long(value[0] != 0));
					break;
				case INT2_OID:
					builder.push(Long(readType<short>(value)));
					break;
				case INT4_OID:
					builder.push(Long(readType<Int>(value)));
					break;
				case INT8_OID:
					builder.push(readType<Long>(value));
					break;
				case FLOAT4_OID:
					builder.push(readType<Float>(value));
					break;
				case FLOAT8_OID:
					builder.push(readType<Double>(value));
					break;
				default:
					throw new (this) SQLError(TO_S(this, S("Unsupported column type: ") << type));
				}
			}

			checkAndClear(result);
			return Maybe<Row>(Row(builder));
		} catch (...) {
			checkAndClear(result);
			throw;
		}
	}

	void PostgreSQL::Stmt::disposeResultSeq() {
		const PGDriver &d = driver();
		PostgreSQL *o = owner();

		// If we are currently streaming data, try to cancel the current operation.
		if (streamingData) {
			// For wide support across versions, we support two interfaces here.
			if (o->newCancel) {
				(*d.PQcancelBlocking)(o->newCancel);
			} else if (o->oldCancel) {
				char errbuf[256];
				if ((*d.PQcancel)(o->oldCancel, errbuf, sizeof(errbuf)) == 0) {
					// Failed to cancel, we more or less ignore that since we can just consume the
					// results anyway. A valid failure is "you cancelled too late", for example.
					// Regardless, we still need to read results until no more are available.
				}
			}
		}

		// We need to call PQgetResult until it returns null.
		PGresult *result = null;
		do {
			result = (*d.PQgetResult)(pgConnection());
			checkAndClear(result);
		} while (result);
		streamingData = false;
	}

	void PostgreSQL::Stmt::checkAndClear(PGresult *result) {
		owner()->checkAndClear(result);
	}

	/**
	 * Visitor.
	 */

	PostgreSQL::Visitor::Visitor() : placeholderCount(0) {}

	void PostgreSQL::Visitor::name(StrBuf *to, Str *name) {
		*to << S("\"") << name << S("\"");
	}

	void PostgreSQL::Visitor::placeholder(StrBuf *to) {
		*to << S("$") << ++placeholderCount;
	}

	void PostgreSQL::Visitor::type(StrBuf *to, QueryType type) {
		Maybe<Nat> size = type.size();

		if (type.sameType(QueryType::text())) {
			if (size.any()) {
				*to << S("VARCHAR") << S("(") << size.value() << S(")");
			} else {
				*to << S("TEXT");
			}
		} else if (type.sameType(QueryType::integer())) {
			if (size.any()) {
				if (size.value() <= 2)
					*to << S("SMALLINT");
				else if (size.value() > 4)
					*to << S("BIGINT");
				else
					*to << S("INTEGER");
			} else {
				*to << S("INTEGER");
			}
		} else if (type.sameType(QueryType::real())) {
			if (size.any() && size.value() > 4) {
				*to << S("DOUBLE PRECISION");
			} else {
				*to << S("REAL");
			}
		} else {
			throw new (this) SQLError(TO_S(this, S("Unsupported type: ") << type << S(".")));
		}
	}

	void PostgreSQL::Visitor::autoIncrement(StrBuf *to) {
		*to << S("GENERATED BY DEFAULT AS IDENTITY");
	}

	void PostgreSQL::Visitor::lastRowId(StrBuf *to) {
		*to << S("lastval()");
	}

}