data_node.c

/*
 * This file and its contents are licensed under the Timescale License.
 * Please see the included NOTICE for copyright information and
 * LICENSE-TIMESCALE for a copy of the license.
 */
#include <postgres.h>
#include <access/htup_details.h>
#include <access/xact.h>
#include <catalog/namespace.h>
#include <catalog/pg_database.h>
#include <catalog/pg_foreign_server.h>
#include <catalog/pg_inherits.h>
#include <catalog/pg_namespace.h>
#include <commands/dbcommands.h>
#include <commands/defrem.h>
#include <commands/event_trigger.h>
#include <compat/compat.h>
#include <executor/tuptable.h>
#include <extension.h>
#include <fmgr.h>
#include <funcapi.h>
#include <libpq/crypt.h>
#include <miscadmin.h>
#include <nodes/makefuncs.h>
#include <nodes/parsenodes.h>
#include <nodes/nodes.h>
#include <nodes/value.h>
#include <utils/acl.h>
#include <utils/builtins.h>
#include <utils/array.h>
#include <utils/builtins.h>
#include <utils/guc.h>
#include <utils/inval.h>
#include <utils/palloc.h>
#include <utils/syscache.h>

#include "compat/compat.h"
#include "config.h"
#include "extension.h"
#include "cache.h"
#include "chunk.h"
#include "fdw/fdw.h"
#include "remote/async.h"
#include "remote/connection.h"
#include "remote/connection_cache.h"
#include "remote/dist_commands.h"
#include "data_node.h"
#include "remote/utils.h"
#include "hypertable_cache.h"
#include "errors.h"
#include "dist_util.h"
#include "utils/uuid.h"
#include "mb/pg_wchar.h"
#include "scan_iterator.h"
#include "ts_catalog/catalog.h"
#include "ts_catalog/chunk_data_node.h"
#include "ts_catalog/dimension_partition.h"
#include "ts_catalog/hypertable_data_node.h"

#define TS_DEFAULT_POSTGRES_PORT 5432
#define TS_DEFAULT_POSTGRES_HOST "localhost"

#define ERRCODE_DUPLICATE_DATABASE_STR "42P04"
#define ERRCODE_DUPLICATE_SCHEMA_STR "42P06"

typedef struct DbInfo
{
	NameData name;
	int32 encoding;
	const char *chartype;
	const char *collation;
} DbInfo;

/* A list of databases we try to connect to when bootstrapping a data node */
static const char *bootstrap_databases[] = { "postgres", "template1", "defaultdb" };

static bool data_node_validate_database(TSConnection *conn, const DbInfo *database);

/*
 * get_database_info - given a database OID, look up info about the database
 *
 * Returns:
 *  True if a record for the OID was found, false otherwise.
 */
static bool
get_database_info(Oid dbid, DbInfo *database)
{
	HeapTuple dbtuple;
	Form_pg_database dbrecord;

	dbtuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(dbid));

	if (!HeapTupleIsValid(dbtuple))
		return false;

	dbrecord = (Form_pg_database) GETSTRUCT(dbtuple);

	database->encoding = dbrecord->encoding;

#if PG15_LT
	database->collation = NameStr(dbrecord->datcollate);
	database->chartype = NameStr(dbrecord->datctype);
#else
	/*
	 * Since datcollate and datctype are varlen fields in PG15+ we cannot rely
	 * on GETSTRUCT filling them in as GETSTRUCT only works for fixed-length
	 * non-NULLABLE columns.
	 */
	Datum datum;
	bool isnull;

	datum = SysCacheGetAttr(DATABASEOID, dbtuple, Anum_pg_database_datcollate, &isnull);
	Assert(!isnull);
	database->collation = TextDatumGetCString(datum);

	datum = SysCacheGetAttr(DATABASEOID, dbtuple, Anum_pg_database_datctype, &isnull);
	Assert(!isnull);
	database->chartype = TextDatumGetCString(datum);
#endif

	database->collation = pstrdup(database->collation);
	database->chartype = pstrdup(database->chartype);

	ReleaseSysCache(dbtuple);
	return true;
}

/*
 * Verify that server is TimescaleDB data node and perform optional ACL check.
 *
 * The function returns true if the server is valid TimescaleDB data node and
 * the ACL check succeeds. Otherwise, false is returned, or, an error is thrown
 * if fail_on_aclcheck is set to true.
 */
static bool
validate_foreign_server(const ForeignServer *server, AclMode const mode, bool fail_on_aclcheck)
{
	Oid const fdwid = get_foreign_data_wrapper_oid(EXTENSION_FDW_NAME, false);
	Oid curuserid = GetUserId();
	AclResult aclresult;
	bool valid;

	Assert(NULL != server);
	if (server->fdwid != fdwid)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("data node \"%s\" is not a TimescaleDB server", server->servername)));

	if (mode == ACL_NO_CHECK)
		return true;

	/* Must have permissions on the server object */
	aclresult = pg_foreign_server_aclcheck(server->serverid, curuserid, mode);

	valid = (aclresult == ACLCHECK_OK);

	if (!valid && fail_on_aclcheck)
		aclcheck_error(aclresult, OBJECT_FOREIGN_SERVER, server->servername);

	return valid;
}

/*
 * Lookup the foreign server by name
 */
ForeignServer *
data_node_get_foreign_server(const char *node_name, AclMode mode, bool fail_on_aclcheck,
							 bool missing_ok)
{
	ForeignServer *server;
	bool valid;

	if (node_name == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("data node name cannot be NULL")));

	server = GetForeignServerByName(node_name, missing_ok);
	if (NULL == server)
		return NULL;

	valid = validate_foreign_server(server, mode, fail_on_aclcheck);

	if (mode != ACL_NO_CHECK && !valid)
		return NULL;

	return server;
}

ForeignServer *
data_node_get_foreign_server_by_oid(Oid server_oid, AclMode mode)
{
	ForeignServer *server = GetForeignServer(server_oid);
	bool PG_USED_FOR_ASSERTS_ONLY valid = validate_foreign_server(server, mode, true);
	Assert(valid); /* Should always be valid since we should see error otherwise */
	return server;
}

/*
 * Create a foreign server.
 *
 * Returns true if the server was created and set the `oid` to the server oid.
 */
static bool
create_foreign_server(const char *const node_name, const char *const host, int32 port,
					  const char *const dbname, bool if_not_exists)
{
	ForeignServer *server = NULL;
	ObjectAddress objaddr;
	CreateForeignServerStmt stmt = {
		.type = T_CreateForeignServerStmt,
		.servername = (char *) node_name,
		.fdwname = EXTENSION_FDW_NAME,
		.options = list_make3(makeDefElem("host", (Node *) makeString(pstrdup(host)), -1),
							  makeDefElem("port", (Node *) makeInteger(port), -1),
							  makeDefElem("dbname", (Node *) makeString(pstrdup(dbname)), -1)),
		.if_not_exists = if_not_exists,
	};

	if (NULL == host)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 (errmsg("invalid host"),
				  (errhint("A hostname or IP address must be specified when "
						   "a data node does not already exist.")))));

	if (if_not_exists)
	{
		server = data_node_get_foreign_server(node_name, ACL_NO_CHECK, false, true);

		if (NULL != server)
		{
			ereport(NOTICE,
					(errcode(ERRCODE_DUPLICATE_OBJECT),
					 errmsg("data node \"%s\" already exists, skipping", node_name)));
			return false;
		}
	}

	/* Permissions checks done in CreateForeignServer() */
	objaddr = CreateForeignServer(&stmt);

	/* CreateForeignServer returns InvalidOid if server already exists */
	if (!OidIsValid(objaddr.objectId))
	{
		Assert(if_not_exists);
		return false;
	}

	return true;
}

TSConnection *
data_node_get_connection(const char *const data_node, RemoteTxnPrepStmtOption const ps_opt,
						 bool transactional)
{
	const ForeignServer *server;
	TSConnectionId id;

	server = data_node_get_foreign_server(data_node, ACL_NO_CHECK, false, false);
	id = remote_connection_id(server->serverid, GetUserId());

	if (transactional)
		return remote_dist_txn_get_connection(id, ps_opt);

	return remote_connection_cache_get_connection(id);
}

/* Attribute numbers for datum returned by create_data_node() */
enum Anum_create_data_node
{
	Anum_create_data_node_name = 1,
	Anum_create_data_node_host,
	Anum_create_data_node_port,
	Anum_create_data_node_dbname,
	Anum_create_data_node_node_created,
	Anum_create_data_node_database_created,
	Anum_create_data_node_extension_created,
	_Anum_create_data_node_max,
};

#define Natts_create_data_node (_Anum_create_data_node_max - 1)

static Datum
create_data_node_datum(FunctionCallInfo fcinfo, const char *node_name, const char *host, int32 port,
					   const char *dbname, bool node_created, bool database_created,
					   bool extension_created)
{
	TupleDesc tupdesc;
	Datum values[_Anum_create_data_node_max];
	bool nulls[_Anum_create_data_node_max] = { false };
	HeapTuple tuple;

	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("function returning record called in "
						"context that cannot accept type record")));

	tupdesc = BlessTupleDesc(tupdesc);
	values[AttrNumberGetAttrOffset(Anum_create_data_node_name)] = CStringGetDatum(node_name);
	values[AttrNumberGetAttrOffset(Anum_create_data_node_host)] = CStringGetTextDatum(host);
	values[AttrNumberGetAttrOffset(Anum_create_data_node_port)] = Int32GetDatum(port);
	values[AttrNumberGetAttrOffset(Anum_create_data_node_dbname)] = CStringGetDatum(dbname);
	values[AttrNumberGetAttrOffset(Anum_create_data_node_node_created)] =
		BoolGetDatum(node_created);
	values[AttrNumberGetAttrOffset(Anum_create_data_node_database_created)] =
		BoolGetDatum(database_created);
	values[AttrNumberGetAttrOffset(Anum_create_data_node_extension_created)] =
		BoolGetDatum(extension_created);
	tuple = heap_form_tuple(tupdesc, values, nulls);

	return HeapTupleGetDatum(tuple);
}

static Datum
create_hypertable_data_node_datum(FunctionCallInfo fcinfo, HypertableDataNode *node)
{
	TupleDesc tupdesc;
	Datum values[Natts_hypertable_data_node];
	bool nulls[Natts_hypertable_data_node] = { false };
	HeapTuple tuple;

	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("function returning record called in "
						"context that cannot accept type record")));

	tupdesc = BlessTupleDesc(tupdesc);
	values[AttrNumberGetAttrOffset(Anum_hypertable_data_node_hypertable_id)] =
		Int32GetDatum(node->fd.hypertable_id);
	values[AttrNumberGetAttrOffset(Anum_hypertable_data_node_node_hypertable_id)] =
		Int32GetDatum(node->fd.node_hypertable_id);
	values[AttrNumberGetAttrOffset(Anum_hypertable_data_node_node_name)] =
		NameGetDatum(&node->fd.node_name);
	tuple = heap_form_tuple(tupdesc, values, nulls);

	return HeapTupleGetDatum(tuple);
}

static List *
create_data_node_options(const char *host, int32 port, const char *dbname, const char *user,
						 const char *password)
{
	DefElem *host_elm = makeDefElem("host", (Node *) makeString(pstrdup(host)), -1);
	DefElem *port_elm = makeDefElem("port", (Node *) makeInteger(port), -1);
	DefElem *dbname_elm = makeDefElem("dbname", (Node *) makeString(pstrdup(dbname)), -1);
	DefElem *user_elm = makeDefElem("user", (Node *) makeString(pstrdup(user)), -1);

	if (NULL != password)
	{
		DefElem *password_elm = makeDefElem("password", (Node *) makeString(pstrdup(password)), -1);
		return list_make5(host_elm, port_elm, dbname_elm, user_elm, password_elm);
	}

	return list_make4(host_elm, port_elm, dbname_elm, user_elm);
}

/* Returns 'true' if the database was created. */
static bool
data_node_bootstrap_database(TSConnection *conn, const DbInfo *database)
{
	const char *const username = PQuser(remote_connection_get_pg_conn(conn));

	Assert(database);

	if (data_node_validate_database(conn, database))
	{
		/* If the database already existed on the remote node, we will log a
		 * notice and proceed since it is not an error if the database already
		 * existed on the remote node. */
		elog(NOTICE,
			 "database \"%s\" already exists on data node, skipping",
			 NameStr(database->name));
		return false;
	}

	/* Create the database with the user as owner. There is no need to
	 * validate the database after this command since it should be created
	 * correctly. */
	PGresult *res =
		remote_connection_execf(conn,
								"CREATE DATABASE %s ENCODING %s LC_COLLATE %s LC_CTYPE %s "
								"TEMPLATE template0 OWNER %s",
								quote_identifier(NameStr(database->name)),
								quote_identifier(pg_encoding_to_char(database->encoding)),
								quote_literal_cstr(database->collation),
								quote_literal_cstr(database->chartype),
								quote_identifier(username));
	if (PQresultStatus(res) != PGRES_COMMAND_OK)
		remote_result_elog(res, ERROR);
	return true;
}

/* Validate the database.
 *
 * Errors:
 *   Will abort with errors if the database exists but is not correctly set
 *   up.
 * Returns:
 *   true if the database exists and is valid
 *   false if it does not exist.
 */
static bool
data_node_validate_database(TSConnection *conn, const DbInfo *database)
{
	PGresult *res;
	int32 actual_encoding;
	const char *actual_chartype;
	const char *actual_collation;

	res = remote_connection_execf(conn,
								  "SELECT encoding, datcollate, datctype "
								  "FROM pg_database WHERE datname = %s",
								  quote_literal_cstr(NameStr(database->name)));

	if (PQresultStatus(res) != PGRES_TUPLES_OK)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_EXCEPTION), errmsg("%s", PQresultErrorMessage(res))));

	if (PQntuples(res) == 0)
		return false;

	Assert(PQnfields(res) > 2);

	actual_encoding = atoi(PQgetvalue(res, 0, 0));
	if (actual_encoding != database->encoding)
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_INVALID_CONFIG),
				 errmsg("database exists but has wrong encoding"),
				 errdetail("Expected database encoding to be \"%s\" (%u) but it was \"%s\" (%u).",
						   pg_encoding_to_char(database->encoding),
						   database->encoding,
						   pg_encoding_to_char(actual_encoding),
						   actual_encoding)));

	actual_collation = PQgetvalue(res, 0, 1);
	Assert(actual_collation != NULL);
	if (strcmp(actual_collation, database->collation) != 0)
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_INVALID_CONFIG),
				 errmsg("database exists but has wrong collation"),
				 errdetail("Expected collation \"%s\" but it was \"%s\".",
						   database->collation,
						   actual_collation)));

	actual_chartype = PQgetvalue(res, 0, 2);
	Assert(actual_chartype != NULL);
	if (strcmp(actual_chartype, database->chartype) != 0)
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_INVALID_CONFIG),
				 errmsg("database exists but has wrong LC_CTYPE"),
				 errdetail("Expected LC_CTYPE \"%s\" but it was \"%s\".",
						   database->chartype,
						   actual_chartype)));
	return true;
}

static void
data_node_validate_extension(TSConnection *conn)
{
	const char *const dbname = PQdb(remote_connection_get_pg_conn(conn));
	const char *const host = PQhost(remote_connection_get_pg_conn(conn));
	const char *const port = PQport(remote_connection_get_pg_conn(conn));

	if (!remote_connection_check_extension(conn))
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_INVALID_CONFIG),
				 errmsg("database does not have TimescaleDB extension loaded"),
				 errdetail("The TimescaleDB extension is not loaded in database %s on node at "
						   "%s:%s.",
						   dbname,
						   host,
						   port)));
}

static void
data_node_validate_as_data_node(TSConnection *conn)
{
	PGresult *res =
		remote_connection_exec(conn, "SELECT _timescaledb_internal.validate_as_data_node()");

	if (PQresultStatus(res) != PGRES_TUPLES_OK)
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_INVALID_CONFIG),
				 (errmsg("cannot add \"%s\" as a data node", remote_connection_node_name(conn)),
				  errdetail("%s", PQresultErrorMessage(res)))));

	remote_result_close(res);
}

/*
 * Bootstrap the extension and associated objects.
 */
static bool
data_node_bootstrap_extension(TSConnection *conn)
{
	const char *const username = PQuser(remote_connection_get_pg_conn(conn));
	const char *schema_name = ts_extension_schema_name();
	const char *schema_name_quoted = quote_identifier(schema_name);
	Oid schema_oid = get_namespace_oid(schema_name, true);

	/* We only count the number of tuples in the code below, but having the
	 * name and version are useful for debugging purposes. */
	PGresult *res =
		remote_connection_execf(conn,
								"SELECT extname, extversion FROM pg_extension WHERE extname = %s",
								quote_literal_cstr(EXTENSION_NAME));

	if (PQresultStatus(res) != PGRES_TUPLES_OK)
		remote_result_elog(res, ERROR);

	if (PQntuples(res) == 0)
	{
		remote_result_close(res);

		if (schema_oid != PG_PUBLIC_NAMESPACE)
		{
			res = remote_connection_execf(conn,
										  "CREATE SCHEMA %s AUTHORIZATION %s",
										  schema_name_quoted,
										  quote_identifier(username));

			if (PQresultStatus(res) != PGRES_COMMAND_OK)
			{
				const char *const sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE);
				bool schema_exists =
					(sqlstate && strcmp(sqlstate, ERRCODE_DUPLICATE_SCHEMA_STR) == 0);
				if (!schema_exists)
					remote_result_elog(res, ERROR);

				remote_result_close(res);
				/* If the schema already existed on the remote node, we got a
				 * duplicate schema error and the schema was not created. In
				 * that case, we log an error with a hint on how to fix the
				 * issue. */
				ereport(ERROR,
						(errcode(ERRCODE_DUPLICATE_SCHEMA),
						 errmsg("schema \"%s\" already exists in database, aborting", schema_name),
						 errhint("Make sure that the data node does not contain any "
								 "existing objects prior to adding it.")));
			}

			remote_result_close(res);
		}

		remote_connection_cmdf_ok(conn,
								  "CREATE EXTENSION " EXTENSION_NAME
								  " WITH SCHEMA %s VERSION %s CASCADE",
								  schema_name_quoted,
								  quote_literal_cstr(ts_extension_get_version()));
		return true;
	}
	else
	{
		ereport(NOTICE,
				(errmsg("extension \"%s\" already exists on data node, skipping",
						PQgetvalue(res, 0, 0)),
				 errdetail("TimescaleDB extension version on %s:%s was %s.",
						   PQhost(remote_connection_get_pg_conn(conn)),
						   PQport(remote_connection_get_pg_conn(conn)),
						   PQgetvalue(res, 0, 1))));
		remote_result_close(res);
		data_node_validate_extension(conn);
		return false;
	}
}

/* Add dist_uuid on the remote node.
 *
 * If the remote node is set to use the current database, `set_dist_id` will report an error and not
 * set it. */
static void
add_distributed_id_to_data_node(TSConnection *conn)
{
	Datum id_string = DirectFunctionCall1(uuid_out, dist_util_get_id());
	PGresult *res = remote_connection_queryf_ok(conn,
												"SELECT _timescaledb_internal.set_dist_id('%s')",
												DatumGetCString(id_string));
	remote_result_close(res);
}

/*
 * Connect to do bootstrapping.
 *
 * We iterate through the list of databases and try to connect to so we can
 * bootstrap the data node.
 */
static TSConnection *
connect_for_bootstrapping(const char *node_name, const char *const host, int32 port,
						  const char *username, const char *password)
{
	TSConnection *conn = NULL;
	char *err = NULL;

	for (size_t i = 0; i < lengthof(bootstrap_databases); i++)
	{
		List *node_options =
			create_data_node_options(host, port, bootstrap_databases[i], username, password);
		conn = remote_connection_open(node_name, node_options, TS_NO_TIMEOUT, &err);

		if (conn)
			return conn;
	}

	ereport(ERROR,
			(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
			 errmsg("could not connect to \"%s\"", node_name),
			 err == NULL ? 0 : errdetail("%s", err)));

	pg_unreachable();

	return NULL;
}

/*
 * Validate that compatible extension is available on the data node.
 *
 * We check all available extension versions. Since we are connected to
 * template DB when performing this check, it means we can't
 * really tell if a compatible extension is installed in the database we
 * are trying to add to the cluster. However we can make sure that a user
 * will be able to manually upgrade the extension on the data node if needed.
 *
 * Will abort with error if there is no compatible version available, otherwise do nothing.
 */
static void
data_node_validate_extension_availability(TSConnection *conn)
{
	StringInfo concat_versions = makeStringInfo();
	bool compatible = false;
	PGresult *res;
	int i;

	res =
		remote_connection_execf(conn,
								"SELECT version FROM pg_available_extension_versions WHERE name = "
								"%s AND version ~ '\\d+.\\d+.\\d+.*' ORDER BY version DESC",
								quote_literal_cstr(EXTENSION_NAME));

	if (PQresultStatus(res) != PGRES_TUPLES_OK)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_EXCEPTION),
				 errmsg("failed to validate remote extension: %s", PQresultErrorMessage(res))));

	if (PQntuples(res) == 0)
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_INVALID_CONFIG),
				 errmsg("TimescaleDB extension not available on remote PostgreSQL instance"),
				 errhint("Install the TimescaleDB extension on the remote PostgresSQL instance.")));

	Assert(PQnfields(res) == 1);

	for (i = 0; i < PQntuples(res); i++)
	{
		appendStringInfo(concat_versions, "%s, ", PQgetvalue(res, i, 0));
		compatible = dist_util_is_compatible_version(PQgetvalue(res, i, 0), TIMESCALEDB_VERSION);
		if (compatible)
			break;
	}

	if (!compatible)
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_INVALID_CONFIG),
				 errmsg("remote PostgreSQL instance has an incompatible timescaledb extension "
						"version"),
				 errdetail_internal("Access node version: %s, available remote versions: %s.",
									TIMESCALEDB_VERSION_MOD,
									concat_versions->data)));
}

/**
 * Get the configured server port for the server as an integer.
 *
 * Returns:
 *   Port number if a port is configured, -1 if it is not able to get
 *   the port number.
 *
 * Note:
 *   We cannot use `inet_server_port()` since that will return NULL if
 *   connecting to a server on localhost since a UNIX socket will be
 *   used. This is the case even if explicitly using a port when
 *   connecting. Regardless of how the user connected, we want to use the same
 *   port as the one that the server listens on.
 */
static int32
get_server_port()
{
	const char *const portstr =
		GetConfigOption("port", /* missing_ok= */ false, /* restrict_privileged= */ false);
	return pg_strtoint32(portstr);
}

static void
validate_data_node_port(int port)
{
	if (port < 1 || port > PG_UINT16_MAX)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 (errmsg("invalid port number %d", port),
				  errhint("The port number must be between 1 and %u.", PG_UINT16_MAX))));
}

/* set_distid may need to be false for some otherwise invalid configurations
 * that are useful for testing */
static Datum
data_node_add_internal(PG_FUNCTION_ARGS, bool set_distid)
{
	Oid userid = GetUserId();
	const char *username = GetUserNameFromId(userid, false);
	const char *node_name = PG_ARGISNULL(0) ? NULL : PG_GETARG_CSTRING(0);
	const char *host = PG_ARGISNULL(1) ? NULL : TextDatumGetCString(PG_GETARG_DATUM(1));
	const char *dbname = PG_ARGISNULL(2) ? get_database_name(MyDatabaseId) : PG_GETARG_CSTRING(2);
	int32 port = PG_ARGISNULL(3) ? get_server_port() : PG_GETARG_INT32(3);
	bool if_not_exists = PG_ARGISNULL(4) ? false : PG_GETARG_BOOL(4);
	bool bootstrap = PG_ARGISNULL(5) ? true : PG_GETARG_BOOL(5);
	const char *password = PG_ARGISNULL(6) ? NULL : TextDatumGetCString(PG_GETARG_DATUM(6));
	bool server_created = false;
	bool database_created = false;
	bool extension_created = false;
	bool PG_USED_FOR_ASSERTS_ONLY result;
	DbInfo database;

	TS_PREVENT_FUNC_IF_READ_ONLY();

	namestrcpy(&database.name, dbname);

	if (host == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 (errmsg("a host needs to be specified"),
				  errhint("Provide a host name or IP address of a data node to add."))));

	if (set_distid && dist_util_membership() == DIST_MEMBER_DATA_NODE)
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_ASSIGNMENT_ALREADY_EXISTS),
				 (errmsg("unable to assign data nodes from an existing distributed database"))));

	if (NULL == node_name)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 (errmsg("data node name cannot be NULL"))));

	validate_data_node_port(port);

	result = get_database_info(MyDatabaseId, &database);
	Assert(result);

	/*
	 * Since this function creates databases on remote nodes, and CREATE DATABASE
	 * cannot run in a transaction block, we cannot run the function in a
	 * transaction block either.
	 */
	TS_PREVENT_IN_TRANSACTION_BLOCK(true);

	/* Try to create the foreign server, or get the existing one in case of
	 * if_not_exists true. */
	if (create_foreign_server(node_name, host, port, dbname, if_not_exists))
	{
		List *node_options;
		TSConnection *conn;

		server_created = true;

		/* Make the foreign server visible in current transaction. */
		CommandCounterIncrement();

		/* If bootstrapping, we check the extension availability here and
		 * abort if the extension is not available. We should not start
		 * creating databases and other cruft on the datanode unless we know
		 * that the extension is installed.
		 *
		 * We ensure that there is a database if we are bootstrapping. This is
		 * done using a separate connection since the database that is going
		 * to be used for the data node does not exist yet, so we cannot
		 * connect to it. */
		if (bootstrap)
		{
			TSConnection *conn =
				connect_for_bootstrapping(node_name, host, port, username, password);
			Assert(NULL != conn);
			data_node_validate_extension_availability(conn);
			database_created = data_node_bootstrap_database(conn, &database);
			remote_connection_close(conn);
		}

		/* Connect to the database we are bootstrapping and either install the
		 * extension or validate that the extension is installed. The
		 * following statements are executed inside a transaction so that they
		 * can be rolled back in the event of a failure.
		 *
		 * We could use `remote_dist_txn_get_connection` here, but it is
		 * comparably heavy and make the code more complicated than
		 * necessary. Instead using a more straightforward approach here since
		 * we do not need 2PC support. */
		node_options = create_data_node_options(host, port, dbname, username, password);
		conn = remote_connection_open_session(node_name, node_options, false);
		Assert(NULL != conn);
		remote_connection_cmd_ok(conn, "BEGIN");

		if (bootstrap)
			extension_created = data_node_bootstrap_extension(conn);

		if (!database_created)
		{
			data_node_validate_database(conn, &database);
			data_node_validate_as_data_node(conn);
		}

		if (!extension_created)
			data_node_validate_extension(conn);

		/* After the node is verified or bootstrapped, we set the `dist_uuid`
		 * using the same connection. We skip this if clustering checks are
		 * disabled, which means that the `dist_uuid` is neither set nor
		 * checked.
		 *
		 * This is done inside a transaction so that we can roll it back if
		 * there are any failures. Note that any failure at this point will
		 * not rollback the creates above. */
		if (set_distid)
		{
			if (dist_util_membership() != DIST_MEMBER_ACCESS_NODE)
				dist_util_set_as_access_node();
			add_distributed_id_to_data_node(conn);
		}

		/* If there were an error before, we will not reach this point to the
		 * transaction will be aborted when the connection is closed. */
		remote_connection_cmd_ok(conn, "COMMIT");
		remote_connection_close(conn);
	}

	PG_RETURN_DATUM(create_data_node_datum(fcinfo,
										   node_name,
										   host,
										   port,
										   dbname,
										   server_created,
										   database_created,
										   extension_created));
}

Datum
data_node_add(PG_FUNCTION_ARGS)
{
	return data_node_add_internal(fcinfo, true);
}

Datum
data_node_add_without_dist_id(PG_FUNCTION_ARGS)
{
	return data_node_add_internal(fcinfo, false);
}

Datum
data_node_attach(PG_FUNCTION_ARGS)
{
	const char *node_name = PG_ARGISNULL(0) ? NULL : PG_GETARG_CSTRING(0);
	Oid table_id = PG_GETARG_OID(1);
	bool if_not_attached = PG_ARGISNULL(2) ? false : PG_GETARG_BOOL(2);
	bool repartition = PG_ARGISNULL(3) ? false : PG_GETARG_BOOL(3);
	ForeignServer *fserver;
	HypertableDataNode *node;
	Cache *hcache;
	Hypertable *ht;
	Dimension *space_dim;
	List *result;
	int num_nodes;
	ListCell *lc;
	Oid uid, saved_uid;
	int sec_ctx;
	Relation rel;

	TS_PREVENT_FUNC_IF_READ_ONLY();

	if (PG_ARGISNULL(1))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("hypertable cannot be NULL")));
	Assert(get_rel_name(table_id));

	ht = ts_hypertable_cache_get_cache_and_entry(table_id, CACHE_FLAG_NONE, &hcache);
	Assert(ht != NULL);

	if (!hypertable_is_distributed(ht))
		ereport(ERROR,
				(errcode(ERRCODE_TS_HYPERTABLE_NOT_DISTRIBUTED),
				 errmsg("hypertable \"%s\" is not distributed", get_rel_name(table_id))));

	/* Must have owner permissions on the hypertable to attach a new data
	   node. Must also have USAGE on the foreign server.  */
	ts_hypertable_permissions_check(table_id, GetUserId());
	fserver = data_node_get_foreign_server(node_name, ACL_USAGE, true, false);

	Assert(NULL != fserver);

	foreach (lc, ht->data_nodes)
	{
		node = lfirst(lc);

		if (node->foreign_server_oid == fserver->serverid)
		{
			ts_cache_release(hcache);

			if (if_not_attached)
			{
				ereport(NOTICE,
						(errcode(ERRCODE_TS_DATA_NODE_ALREADY_ATTACHED),
						 errmsg("data node \"%s\" is already attached to hypertable \"%s\", "
								"skipping",
								node_name,
								get_rel_name(table_id))));
				PG_RETURN_DATUM(create_hypertable_data_node_datum(fcinfo, node));
			}
			else
				ereport(ERROR,
						(errcode(ERRCODE_TS_DATA_NODE_ALREADY_ATTACHED),
						 errmsg("data node \"%s\" is already attached to hypertable \"%s\"",
								node_name,
								get_rel_name(table_id))));
		}
	}

	/*
	 * Change to the hypertable owner so that the same permissions will be set up on the
	 * datanode being attached to as well. We need to do this explicitly because the
	 * caller of this function could be a superuser and we definitely don't want to create
	 * this hypertable with superuser ownership on the datanode being attached to!
	 *
	 * We retain the lock on the hypertable till the end of the traction to avoid any
	 * possibility of a concurrent "ALTER TABLE OWNER TO" changing the owner underneath
	 * us.
	 */
	rel = table_open(ht->main_table_relid, AccessShareLock);
	uid = rel->rd_rel->relowner;
	table_close(rel, NoLock);
	GetUserIdAndSecContext(&saved_uid, &sec_ctx);

	if (uid != saved_uid)
		SetUserIdAndSecContext(uid, sec_ctx | SECURITY_LOCAL_USERID_CHANGE);

	result = hypertable_assign_data_nodes(ht->fd.id, list_make1((char *) node_name));
	Assert(result->length == 1);

	/* Refresh the cached hypertable entry to get the attached node */
	ts_cache_release(hcache);
	hcache = ts_hypertable_cache_pin();
	ht = ts_hypertable_cache_get_entry(hcache, table_id, CACHE_FLAG_NONE);

	/* Get the first closed (space) dimension, which is the one along which we
	 * partition across data nodes. */
	space_dim = ts_hyperspace_get_mutable_dimension(ht->space, DIMENSION_TYPE_CLOSED, 0);
	num_nodes = list_length(ht->data_nodes);

	if (num_nodes > MAX_NUM_HYPERTABLE_DATA_NODES)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("max number of data nodes already attached"),
				 errdetail("The number of data nodes in a hypertable cannot exceed %d.",
						   MAX_NUM_HYPERTABLE_DATA_NODES)));

	/* If there are less slices (partitions) in the space dimension than there
	 * are data nodes, we'd like to expand the number of slices to be able to
	 * make use of the new data node. */
	if (NULL != space_dim)
	{
		List *data_node_names = NIL;
		int num_partitions = space_dim->fd.num_slices;

		if (num_nodes > space_dim->fd.num_slices)
		{
			if (repartition)
			{
				ts_dimension_set_number_of_slices(space_dim, num_nodes & 0xFFFF);
				num_partitions = num_nodes;

				ereport(NOTICE,
						(errmsg("the number of partitions in dimension \"%s\" was increased to %u",
								NameStr(space_dim->fd.column_name),
								num_nodes),
						 errdetail("To make use of all attached data nodes, a distributed "
								   "hypertable needs at least as many partitions in the first "
								   "closed (space) dimension as there are attached data nodes.")));
			}
			else
			{
				/* Raise a warning if the number of partitions are too few to make
				 * use of all data nodes. Need to refresh cache first to get the
				 * updated data node list. */
				int dimension_id = space_dim->fd.id;
				ts_hypertable_check_partitioning(ht, dimension_id);
			}
		}

		data_node_names = ts_hypertable_get_available_data_node_names(ht, true);
		ts_dimension_partition_info_recreate(space_dim->fd.id,
											 num_partitions,
											 data_node_names,
											 ht->fd.replication_factor);
	}

	node = linitial(result);
	ts_cache_release(hcache);

	/* Need to restore security context */
	if (uid != saved_uid)
		SetUserIdAndSecContext(saved_uid, sec_ctx);

	PG_RETURN_DATUM(create_hypertable_data_node_datum(fcinfo, node));
}

/* Only used for generating proper error message */
typedef enum OperationType
{
	OP_BLOCK,
	OP_DETACH,
	OP_DELETE
} OperationType;

static void
check_replication_for_new_data(const char *node_name, Hypertable *ht, bool force)
{
	List *available_nodes = ts_hypertable_get_available_data_nodes(ht, false);

	if (ht->fd.replication_factor < list_length(available_nodes))
		return;

	ereport(force ? WARNING : ERROR,
			(errcode(ERRCODE_TS_INSUFFICIENT_NUM_DATA_NODES),
			 errmsg("insufficient number of data nodes for distributed hypertable \"%s\"",
					NameStr(ht->fd.table_name)),
			 errdetail("Reducing the number of available data nodes on distributed"
					   " hypertable \"%s\" prevents full replication of new chunks.",
					   NameStr(ht->fd.table_name)),
			 force ? 0 : errhint("Use force => true to force this operation.")));
}

static bool
data_node_contains_non_replicated_chunks(List *chunk_data_nodes)
{
	ListCell *lc;

	foreach (lc, chunk_data_nodes)
	{
		ChunkDataNode *cdn = lfirst(lc);
		List *replicas =
			ts_chunk_data_node_scan_by_chunk_id(cdn->fd.chunk_id, CurrentMemoryContext);

		if (list_length(replicas) < 2)
			return true;
	}

	return false;
}

static List *
data_node_detach_or_delete_validate(const char *node_name, Hypertable *ht, bool force,
									OperationType op_type)
{
	List *chunk_data_nodes =
		ts_chunk_data_node_scan_by_node_name_and_hypertable_id(node_name,
															   ht->fd.id,
															   CurrentMemoryContext);
	bool has_non_replicated_chunks = data_node_contains_non_replicated_chunks(chunk_data_nodes);

	Assert(op_type == OP_DELETE || op_type == OP_DETACH);

	if (has_non_replicated_chunks)
		ereport(ERROR,
				(errcode(ERRCODE_TS_INSUFFICIENT_NUM_DATA_NODES),
				 errmsg("insufficient number of data nodes"),
				 errdetail("Distributed hypertable \"%s\" would lose data if"
						   " data node \"%s\" is %s.",
						   NameStr(ht->fd.table_name),
						   node_name,
						   (op_type == OP_DELETE) ? "deleted" : "detached"),
				 errhint("Ensure all chunks on the data node are fully replicated before %s it.",
						 (op_type == OP_DELETE) ? "deleting" : "detaching")));

	if (list_length(chunk_data_nodes) > 0)
	{
		if (force)
			ereport(WARNING,
					(errcode(ERRCODE_TS_INSUFFICIENT_NUM_DATA_NODES),
					 errmsg("distributed hypertable \"%s\" is under-replicated",
							NameStr(ht->fd.table_name)),
					 errdetail("Some chunks no longer meet the replication target"
							   " after %s data node \"%s\".",
							   (op_type == OP_DELETE) ? "deleting" : "detaching",
							   node_name)));
		else
			ereport(ERROR,
					(errcode(ERRCODE_TS_DATA_NODE_IN_USE),
					 errmsg("data node \"%s\" still holds data for distributed hypertable \"%s\"",
							node_name,
							NameStr(ht->fd.table_name))));
	}

	check_replication_for_new_data(node_name, ht, force);

	return chunk_data_nodes;
}

static int
data_node_modify_hypertable_data_nodes(const char *node_name, List *hypertable_data_nodes,
									   bool all_hypertables, OperationType op_type,
									   bool block_chunks, bool force, bool repartition,
									   bool drop_remote_data)
{
	Cache *hcache = ts_hypertable_cache_pin();
	ListCell *lc;
	int removed = 0;

	foreach (lc, hypertable_data_nodes)
	{
		HypertableDataNode *node = lfirst(lc);
		Oid relid = ts_hypertable_id_to_relid(node->fd.hypertable_id);
		Hypertable *ht = ts_hypertable_cache_get_entry(hcache, relid, CACHE_FLAG_NONE);
		bool has_privs = ts_hypertable_has_privs_of(relid, GetUserId());
		bool update_dimension_partitions = false;
		Dimension *space_dim;

		Assert(ht != NULL);
		space_dim = ts_hyperspace_get_mutable_dimension(ht->space, DIMENSION_TYPE_CLOSED, 0);

		if (!has_privs)
		{
			/* If the operation is OP_DELETE, we MUST be able to detach the data
			 * node from ALL tables since the foreign server object will be
			 * deleted. Therefore, we fail the operation if we find a table
			 * that we don't have owner permissions on in this case. */
			if (all_hypertables && op_type != OP_DELETE)
				ereport(NOTICE,
						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
						 errmsg("skipping hypertable \"%s\" due to missing permissions",
								get_rel_name(relid))));
			else
				ereport(ERROR,
						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
						 errmsg("permission denied for hypertable \"%s\"", get_rel_name(relid)),
						 errdetail("The data node is attached to hypertables that the current "
								   "user lacks permissions for.")));
		}
		else if (op_type == OP_DETACH || op_type == OP_DELETE)
		{
			/* we have permissions to detach */
			List *chunk_data_nodes =
				data_node_detach_or_delete_validate(NameStr(node->fd.node_name),
													ht,
													force,
													op_type);
			ListCell *cs_lc;

			/* update chunk foreign table server and delete chunk mapping */
			foreach (cs_lc, chunk_data_nodes)
			{
				ChunkDataNode *cdn = lfirst(cs_lc);
				const Chunk *chunk = ts_chunk_get_by_id(cdn->fd.chunk_id, true);
				chunk_update_foreign_server_if_needed(chunk, cdn->foreign_server_oid, false);
				ts_chunk_data_node_delete_by_chunk_id_and_node_name(cdn->fd.chunk_id,
																	NameStr(cdn->fd.node_name));
			}

			/* delete hypertable mapping */
			removed +=
				ts_hypertable_data_node_delete_by_node_name_and_hypertable_id(node_name, ht->fd.id);

			if (repartition)
			{
				int num_nodes = list_length(ht->data_nodes) - 1;

				if (space_dim != NULL && num_nodes < space_dim->fd.num_slices && num_nodes > 0)
				{
					ts_dimension_set_number_of_slices(space_dim, num_nodes & 0xFFFF);

					ereport(NOTICE,
							(errmsg("the number of partitions in dimension \"%s\" of hypertable "
									"\"%s\" was decreased to %u",
									NameStr(space_dim->fd.column_name),
									get_rel_name(ht->main_table_relid),
									num_nodes),
							 errdetail(
								 "To make efficient use of all attached data nodes, the number of "
								 "space partitions was set to match the number of data nodes.")));
				}
			}

			/* Update dimension partitions. First remove the detach/deleted
			 * data node from the list of remaining nodes so that it is not
			 * used in the new partitioning scheme.
			 *
			 * Note that the cached dimension partition info in the Dimension
			 * object is not updated. The cache will be invalidated and
			 * released at the end of this function.
			 */
			update_dimension_partitions = NULL != space_dim;

			if (op_type == OP_DETACH && drop_remote_data)
			{
				/* Drop the hypertable on the data node */
				ts_dist_cmd_run_on_data_nodes(
					psprintf("DROP TABLE IF EXISTS %s",
							 quote_qualified_identifier(NameStr(ht->fd.schema_name),
														NameStr(ht->fd.table_name))),
					list_make1(NameStr(node->fd.node_name)),
					true);
			}
		}
		else
		{
			/*  set block new chunks */
			if (block_chunks)
			{
				if (node->fd.block_chunks)
				{
					elog(NOTICE,
						 "new chunks already blocked on data node \"%s\" for"
						 " hypertable \"%s\"",
						 NameStr(node->fd.node_name),
						 get_rel_name(relid));
					continue;
				}

				check_replication_for_new_data(node_name, ht, force);
			}
			node->fd.block_chunks = block_chunks;
			removed += ts_hypertable_data_node_update(node);
			update_dimension_partitions = NULL != space_dim;
		}

		if (update_dimension_partitions)
		{
			/* Refresh the cached hypertable to get the updated list of data nodes */
			ts_cache_release(hcache);
			hcache = ts_hypertable_cache_pin();
			ht = ts_hypertable_cache_get_entry(hcache, relid, CACHE_FLAG_NONE);
			ts_hypertable_update_dimension_partitions(ht);
		}
	}

	ts_cache_release(hcache);
	return removed;
}

static int
data_node_block_hypertable_data_nodes(const char *node_name, List *hypertable_data_nodes,
									  bool all_hypertables, bool block_chunks, bool force)
{
	return data_node_modify_hypertable_data_nodes(node_name,
												  hypertable_data_nodes,
												  all_hypertables,
												  OP_BLOCK,
												  block_chunks,
												  force,
												  false,
												  false);
}

static int
data_node_detach_hypertable_data_nodes(const char *node_name, List *hypertable_data_nodes,
									   bool all_hypertables, bool force, bool repartition,
									   bool drop_remote_data, OperationType op_type)
{
	return data_node_modify_hypertable_data_nodes(node_name,
												  hypertable_data_nodes,
												  all_hypertables,
												  op_type,
												  false,
												  force,
												  repartition,
												  drop_remote_data);
}

HypertableDataNode *
data_node_hypertable_get_by_node_name(const Hypertable *ht, const char *node_name,
									  bool attach_check)
{
	HypertableDataNode *hdn = NULL;
	ListCell *lc;

	if (!hypertable_is_distributed(ht))
		ereport(ERROR,
				(errcode(ERRCODE_TS_HYPERTABLE_NOT_DISTRIBUTED),
				 errmsg("hypertable \"%s\" is not distributed",
						get_rel_name(ht->main_table_relid))));

	foreach (lc, ht->data_nodes)
	{
		hdn = lfirst(lc);
		if (namestrcmp(&hdn->fd.node_name, node_name) == 0)
			break;
		else
			hdn = NULL;
	}

	if (hdn == NULL)
	{
		if (attach_check)
			ereport(ERROR,
					(errcode(ERRCODE_TS_DATA_NODE_NOT_ATTACHED),
					 errmsg("data node \"%s\" is not attached to hypertable \"%s\"",
							node_name,
							get_rel_name(ht->main_table_relid))));
		else
			ereport(NOTICE,
					(errcode(ERRCODE_TS_DATA_NODE_NOT_ATTACHED),
					 errmsg("data node \"%s\" is not attached to hypertable \"%s\", "
							"skipping",
							node_name,
							get_rel_name(ht->main_table_relid))));
	}

	return hdn;
}

static HypertableDataNode *
get_hypertable_data_node(Oid table_id, const char *node_name, bool owner_check, bool attach_check)
{
	HypertableDataNode *hdn = NULL;
	Cache *hcache = ts_hypertable_cache_pin();
	const Hypertable *ht = ts_hypertable_cache_get_entry(hcache, table_id, CACHE_FLAG_NONE);

	if (owner_check)
		ts_hypertable_permissions_check(table_id, GetUserId());

	hdn = data_node_hypertable_get_by_node_name(ht, node_name, attach_check);

	ts_cache_release(hcache);

	return hdn;
}

static Datum
data_node_block_or_allow_new_chunks(const char *node_name, Oid const table_id, bool force,
									bool block_chunks)
{
	int affected = 0;
	bool all_hypertables = !OidIsValid(table_id);
	List *hypertable_data_nodes = NIL;
	ForeignServer *server = data_node_get_foreign_server(node_name, ACL_USAGE, true, false);

	Assert(NULL != server);

	if (OidIsValid(table_id))
	{
		/* Early abort on missing hypertable permissions */
		ts_hypertable_permissions_check(table_id, GetUserId());
		hypertable_data_nodes =
			list_make1(get_hypertable_data_node(table_id, server->servername, true, true));
	}
	else
	{
		/* block or allow for all hypertables */
		hypertable_data_nodes =
			ts_hypertable_data_node_scan_by_node_name(server->servername, CurrentMemoryContext);
	}

	affected = data_node_block_hypertable_data_nodes(server->servername,
													 hypertable_data_nodes,
													 all_hypertables,
													 block_chunks,
													 force);
	return Int32GetDatum(affected);
}

Datum
data_node_allow_new_chunks(PG_FUNCTION_ARGS)
{
	const char *node_name = PG_ARGISNULL(0) ? NULL : NameStr(*PG_GETARG_NAME(0));
	Oid table_id = PG_ARGISNULL(1) ? InvalidOid : PG_GETARG_OID(1);

	TS_PREVENT_FUNC_IF_READ_ONLY();

	return data_node_block_or_allow_new_chunks(node_name, table_id, false, false);
}

Datum
data_node_block_new_chunks(PG_FUNCTION_ARGS)
{
	const char *node_name = PG_ARGISNULL(0) ? NULL : NameStr(*PG_GETARG_NAME(0));
	Oid table_id = PG_ARGISNULL(1) ? InvalidOid : PG_GETARG_OID(1);
	bool force = PG_ARGISNULL(2) ? false : PG_GETARG_BOOL(2);

	TS_PREVENT_FUNC_IF_READ_ONLY();

	return data_node_block_or_allow_new_chunks(node_name, table_id, force, true);
}

Datum
data_node_detach(PG_FUNCTION_ARGS)
{
	const char *node_name = PG_ARGISNULL(0) ? NULL : NameStr(*PG_GETARG_NAME(0));
	Oid table_id = PG_ARGISNULL(1) ? InvalidOid : PG_GETARG_OID(1);
	bool all_hypertables = PG_ARGISNULL(1);
	bool if_attached = PG_ARGISNULL(2) ? false : PG_GETARG_BOOL(2);
	bool force = PG_ARGISNULL(3) ? InvalidOid : PG_GETARG_OID(3);
	bool repartition = PG_ARGISNULL(4) ? false : PG_GETARG_BOOL(4);
	bool drop_remote_data = PG_ARGISNULL(5) ? false : PG_GETARG_BOOL(5);
	int removed = 0;
	List *hypertable_data_nodes = NIL;
	ForeignServer *server;

	TS_PREVENT_FUNC_IF_READ_ONLY();

	server = data_node_get_foreign_server(node_name, ACL_USAGE, true, false);
	Assert(NULL != server);

	if (OidIsValid(table_id))
	{
		HypertableDataNode *node;

		/* Early abort on missing hypertable permissions */
		ts_hypertable_permissions_check(table_id, GetUserId());

		node = get_hypertable_data_node(table_id, server->servername, true, !if_attached);
		if (node)
			hypertable_data_nodes = list_make1(node);
	}
	else
	{
		/* Detach data node for all hypertables where user has
		 * permissions. Permissions checks done in
		 * data_node_detach_hypertable_data_nodes().  */
		hypertable_data_nodes =
			ts_hypertable_data_node_scan_by_node_name(server->servername, CurrentMemoryContext);
	}

	removed = data_node_detach_hypertable_data_nodes(server->servername,
													 hypertable_data_nodes,
													 all_hypertables,
													 force,
													 repartition,
													 drop_remote_data,
													 OP_DETACH);

	PG_RETURN_INT32(removed);
}

enum Anum_show_conn
{
	Anum_alter_data_node_node_name = 1,
	Anum_alter_data_node_host,
	Anum_alter_data_node_port,
	Anum_alter_data_node_database,
	Anum_alter_data_node_available,
	_Anum_alter_data_node_max,
};

#define Natts_alter_data_node (_Anum_alter_data_node_max - 1)

static HeapTuple
create_alter_data_node_tuple(TupleDesc tupdesc, const char *node_name, List *options)
{
	Datum values[Natts_alter_data_node];
	bool nulls[Natts_alter_data_node] = { false };
	ListCell *lc;

	MemSet(nulls, false, sizeof(nulls));

	values[AttrNumberGetAttrOffset(Anum_alter_data_node_node_name)] = CStringGetDatum(node_name);
	values[AttrNumberGetAttrOffset(Anum_alter_data_node_available)] = BoolGetDatum(true);

	foreach (lc, options)
	{
		DefElem *elem = lfirst(lc);

		if (strcmp("host", elem->defname) == 0)
		{
			values[AttrNumberGetAttrOffset(Anum_alter_data_node_host)] =
				CStringGetTextDatum(defGetString(elem));
		}
		else if (strcmp("port", elem->defname) == 0)
		{
			int port = atoi(defGetString(elem));
			values[AttrNumberGetAttrOffset(Anum_alter_data_node_port)] = Int32GetDatum(port);
		}
		else if (strcmp("dbname", elem->defname) == 0)
		{
			values[AttrNumberGetAttrOffset(Anum_alter_data_node_database)] =
				CStringGetDatum(defGetString(elem));
		}
		else if (strcmp("available", elem->defname) == 0)
		{
			values[AttrNumberGetAttrOffset(Anum_alter_data_node_available)] =
				BoolGetDatum(defGetBoolean(elem));
		}
	}

	return heap_form_tuple(tupdesc, values, nulls);
}

/*
 * Switch data node to use for queries on chunks.
 *
 * When available=false it will switch from the given data node to another
 * one, but only if the data node is currently used for queries on the chunk.
 *
 * When available=true it will switch to the given data node, if it is
 * "primary" for the chunk (according to the current partitioning
 * configuration).
 */
static void
switch_data_node_on_chunks(const ForeignServer *datanode, bool available)
{
	unsigned int failed_update_count = 0;
	ScanIterator it = ts_chunk_data_nodes_scan_iterator_create(CurrentMemoryContext);
	ts_chunk_data_nodes_scan_iterator_set_node_name(&it, datanode->servername);

	/* Scan for chunks that reference the given data node */
	ts_scanner_foreach(&it)
	{
		TupleTableSlot *slot = ts_scan_iterator_slot(&it);
		bool PG_USED_FOR_ASSERTS_ONLY isnull = false;
		Datum chunk_id = slot_getattr(slot, Anum_chunk_data_node_chunk_id, &isnull);

		Assert(!isnull);

		const Chunk *chunk = ts_chunk_get_by_id(DatumGetInt32(chunk_id), true);
		if (!chunk_update_foreign_server_if_needed(chunk, datanode->serverid, available))
			failed_update_count++;
	}

	if (!available && failed_update_count > 0)
		elog(WARNING, "could not switch data node on %u chunks", failed_update_count);

	ts_scan_iterator_close(&it);
}

/*
 * Append new data node options.
 *
 * When setting options via AlterForeignServer(), the defelem list must
 * account for whether the an option already exists (is set) in the current
 * options or it is newly added. These are different operations on a foreign
 * server.
 *
 * Any options that already exist are purged from the current_options list so
 * that only the options not set or added remains. This list can be merged
 * with the new options to produce the full list of options (new and old).
 */
static List *
append_data_node_option(List *new_options, List **current_options, const char *name, Node *value)
{
	DefElem *elem;
	ListCell *lc;
	bool option_found = false;
#if PG13_LT
	ListCell *prev_lc = NULL;
#endif

	foreach (lc, *current_options)
	{
		elem = lfirst(lc);

		if (strcmp(elem->defname, name) == 0)
		{
			option_found = true;
			/* Remove the option which is replaced so that the remaining
			 * options can be merged later into an updated list */
#if PG13_GE
			*current_options = list_delete_cell(*current_options, lc);
#else
			*current_options = list_delete_cell(*current_options, lc, prev_lc);
#endif
			break;
		}
#if PG13_LT
		prev_lc = lc;
#endif
	}

	elem = makeDefElemExtended(NULL,
							   pstrdup(name),
							   value,
							   option_found ? DEFELEM_SET : DEFELEM_ADD,
							   -1);
	return lappend(new_options, elem);
}

/*
 * Alter a data node.
 *
 * Change the configuration of a data node, including host, port, and
 * database.
 *
 * Can also be used to mark a data node "unavailable", which ensures it is no
 * longer used for reads as long as there are replica chunks on other data
 * nodes to use for reads instead. If it is not possible to fail over all
 * chunks, a warning will be raised.
 */
Datum
data_node_alter(PG_FUNCTION_ARGS)
{
	const char *node_name = PG_ARGISNULL(0) ? NULL : NameStr(*PG_GETARG_NAME(0));
	const char *host = PG_ARGISNULL(1) ? NULL : TextDatumGetCString(PG_GETARG_DATUM(1));
	const char *database = PG_ARGISNULL(2) ? NULL : NameStr(*PG_GETARG_NAME(2));
	int port = PG_ARGISNULL(3) ? -1 : PG_GETARG_INT32(3);
	bool available_is_null = PG_ARGISNULL(4);
	bool available = available_is_null ? true : PG_GETARG_BOOL(4);
	ForeignServer *server = NULL;
	List *current_options = NIL;
	List *options = NIL;
	TupleDesc tupdesc;
	AlterForeignServerStmt alter_server_stmt = {
		.type = T_AlterForeignServerStmt,
		.servername = node_name ? pstrdup(node_name) : NULL,
		.has_version = false,
		.version = NULL,
		.options = NIL,
	};

	TS_PREVENT_FUNC_IF_READ_ONLY();

	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("function returning record called in context "
						"that cannot accept type record")));

	tupdesc = BlessTupleDesc(tupdesc);

	/* Check if a data node with the given name actually exists, or raise an error. */
	server = data_node_get_foreign_server(node_name, ACL_NO_CHECK, false, false /* missing_ok */);

	if (host == NULL && database == NULL && port == -1 && available_is_null)
		PG_RETURN_DATUM(
			HeapTupleGetDatum(create_alter_data_node_tuple(tupdesc, node_name, server->options)));

	current_options = list_copy(server->options);

	if (host != NULL)
		options = append_data_node_option(options,
										  &current_options,
										  "host",
										  (Node *) makeString((char *) host));

	if (database != NULL)
		options = append_data_node_option(options,
										  &current_options,
										  "dbname",
										  (Node *) makeString((char *) database));

	if (port != -1)
	{
		validate_data_node_port(port);
		options =
			append_data_node_option(options, &current_options, "port", (Node *) makeInteger(port));
	}

	if (!available_is_null)
		options = append_data_node_option(options,
										  &current_options,
										  "available",
										  (Node *) makeString(available ? "true" : "false"));

	alter_server_stmt.options = options;
	AlterForeignServer(&alter_server_stmt);

	/* Drop stale chunks on the unavailable data node, if we are going to
	 * make it available again */
	if (!available_is_null && available && !ts_data_node_is_available_by_server(server))
		ts_chunk_drop_stale_chunks(node_name, NULL);

	/* Make changes to the data node (foreign server object) visible so that
	 * the changes are present when we switch "primary" data node on chunks */
	CommandCounterIncrement();

	/* Update the currently used query data node on all affected chunks to
	 * reflect the new status of the data node */
	switch_data_node_on_chunks(server, available);

	/* Add updated options last as they will take precedence over old options
	 * when creating the result tuple. */
	options = list_concat(current_options, options);

	PG_RETURN_DATUM(HeapTupleGetDatum(create_alter_data_node_tuple(tupdesc, node_name, options)));
}

/*
 * Drop a data node's database.
 *
 * To drop the database on the data node, a connection must be made to another
 * database since one cannot drop the database currently connected
 * to. Therefore, we bypass the connection cache and use a "raw" connection to
 * a standard database (e.g., template0 or postgres), similar to how
 * bootstrapping does it.
 *
 * Note that no password is provided on the command line as is done when
 * bootstrapping. Instead, it is assumed that the current user already has a
 * method to authenticate with the remote data node (e.g., via a password
 * file, certificate, or user mapping). This should normally be the case or
 * otherwise the user wouldn't have been able to use the data node.
 *
 * Note that the user that deletes a data node also must be the database owner
 * on the data node. The database will only be dropped if there are no other
 * concurrent connections so all connections must be closed before being able
 * to drop the database.
 */
static void
drop_data_node_database(const ForeignServer *server)
{
	ListCell *lc;
	TSConnection *conn = NULL;
	Oid userid = GetUserId();
	TSConnectionId connid = {
		.server_id = server->serverid,
		.user_id = userid,
	};
	/* Make a copy of the node name since the server pointer will be
	 * updated */
	char *nodename = pstrdup(server->servername);
	char *dbname = NULL;
	char *err = NULL;

	/* Figure out the name of the database that should be dropped */
	foreach (lc, server->options)
	{
		DefElem *d = lfirst(lc);

		if (strcmp(d->defname, "dbname") == 0)
		{
			dbname = defGetString(d);
			break;
		}
	}

	if (NULL == dbname)
	{
		/* This should not happen unless the configuration is messed up */
		ereport(ERROR,
				(errcode(ERRCODE_TS_DATA_NODE_INVALID_CONFIG),
				 errmsg("could not drop the database on data node \"%s\"", nodename),
				 errdetail("The data node configuration lacks the \"dbname\" option.")));
		pg_unreachable();
		return;
	}

	/* Clear potentially cached connection to the data node for the current
	 * session so that it won't block dropping the database */
	remote_connection_cache_remove(connid);

	/* Cannot connect to the database that is being dropped, so try to connect
	 * to a "standard" bootstrap database that we expect to exist on the data
	 * node */
	for (size_t i = 0; i < lengthof(bootstrap_databases); i++)
	{
		List *conn_options;
		DefElem dbname_elem = {
			.type = T_DefElem,
			.defaction = DEFELEM_SET,
			.defname = "dbname",
			.arg = (Node *) makeString(pstrdup(bootstrap_databases[i])),
		};
		AlterForeignServerStmt stmt = {
			.type = T_AlterForeignServerStmt,
			.servername = nodename,
			.has_version = false,
			.options = list_make1(&dbname_elem),
		};

		/*
		 * We assume that the user already has credentials configured to
		 * connect to the data node, e.g., via a user mapping, password file,
		 * or certificate. But in order to easily make use of those
		 * authentication methods, we need to establish the connection using
		 * the standard connection functions to pick up the foreign server
		 * options and associated user mapping (if such a mapping
		 * exists). However, the foreign server configuration references the
		 * database we are trying to drop, so we first need to update the
		 * foreign server definition to use the bootstrap database. */
		AlterForeignServer(&stmt);

		/* Make changes to foreign server database visible */
		CommandCounterIncrement();

		/* Get the updated server definition */
		server = data_node_get_foreign_server(nodename, ACL_USAGE, true, false);
		/* Open a connection to the bootstrap database using the new server options */
		conn_options = remote_connection_prepare_auth_options(server, userid);
		conn = remote_connection_open(nodename, conn_options, TS_NO_TIMEOUT, &err);

		if (NULL != conn)
			break;
	}

	if (NULL != conn)
	{
		/* Do not include FORCE or IF EXISTS options when dropping the
		 * database. Instead, we expect the database to exist, or the user
		 * has to rerun the command without drop_database=>true set. We
		 * don't force removal if there are other connections to the
		 * database out of caution. If the user wants to forcefully remove
		 * the database, they can do it manually. From PG15, the backend
		 * executing the DROP forces all other backends to close all smgr
		 * fds using the ProcSignalBarrier mechanism. To allow this backend
		 * to handle that interrupt, send the DROP request using the async
		 * API. */
		char *cmd;
		AsyncRequest *req;

		cmd = psprintf("DROP DATABASE %s", quote_identifier(dbname));
		req = async_request_send(conn, cmd);
		Assert(NULL != req);
		async_request_wait_ok_result(req);
		remote_connection_close(conn);
		pfree(req);
		pfree(cmd);
	}
	else
		ereport(ERROR,
				(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
				 errmsg("could not connect to data node \"%s\"", nodename),
				 err == NULL ? 0 : errdetail("%s", err)));
}

Datum
data_node_delete(PG_FUNCTION_ARGS)
{
	const char *node_name = PG_ARGISNULL(0) ? NULL : PG_GETARG_CSTRING(0);
	bool if_exists = PG_ARGISNULL(1) ? false : PG_GETARG_BOOL(1);
	bool force = PG_ARGISNULL(2) ? false : PG_GETARG_BOOL(2);
	bool repartition = PG_ARGISNULL(3) ? false : PG_GETARG_BOOL(3);
	bool drop_database = PG_ARGISNULL(4) ? false : PG_GETARG_BOOL(4);
	List *hypertable_data_nodes = NIL;
	DropStmt stmt;
	ObjectAddress address;
	ObjectAddress secondary_object = {
		.classId = InvalidOid,
		.objectId = InvalidOid,
		.objectSubId = 0,
	};
	Node *parsetree = NULL;
	TSConnectionId cid;
	ForeignServer *server;

	TS_PREVENT_FUNC_IF_READ_ONLY();

	/* Need USAGE to detach. Further owner check done when executing the DROP
	 * statement. */
	server = data_node_get_foreign_server(node_name, ACL_USAGE, true, if_exists);

	Assert(server == NULL ? if_exists : true);

	if (NULL == server)
	{
		elog(NOTICE, "data node \"%s\" does not exist, skipping", node_name);
		PG_RETURN_BOOL(false);
	}

	if (drop_database)
	{
		TS_PREVENT_IN_TRANSACTION_BLOCK(true);
	}

	/* close any pending connections */
	remote_connection_id_set(&cid, server->serverid, GetUserId());
	remote_connection_cache_remove(cid);

	/* detach data node */
	hypertable_data_nodes =
		ts_hypertable_data_node_scan_by_node_name(node_name, CurrentMemoryContext);

	data_node_detach_hypertable_data_nodes(node_name,
										   hypertable_data_nodes,
										   true,
										   force,
										   repartition,
										   false,
										   OP_DELETE);

	/* clean up persistent transaction records */
	remote_txn_persistent_record_delete_for_data_node(server->serverid, NULL);

	stmt = (DropStmt){
		.type = T_DropStmt,
		.objects = list_make1(makeString(pstrdup(node_name))),
		.removeType = OBJECT_FOREIGN_SERVER,
		.behavior = DROP_RESTRICT,
		.missing_ok = if_exists,
	};

	parsetree = (Node *) &stmt;

	if (drop_database)
		drop_data_node_database(server);

	/* Make sure event triggers are invoked so that all dropped objects
	 * are collected during a cascading drop. This ensures all dependent
	 * objects get cleaned up. */
	EventTriggerBeginCompleteQuery();

	PG_TRY();
	{
		ObjectAddressSet(address, ForeignServerRelationId, server->serverid);
		EventTriggerDDLCommandStart(parsetree);
		RemoveObjects(&stmt);
		EventTriggerCollectSimpleCommand(address, secondary_object, parsetree);
		EventTriggerSQLDrop(parsetree);
		EventTriggerDDLCommandEnd(parsetree);
	}
	PG_CATCH();
	{
		EventTriggerEndCompleteQuery();
		PG_RE_THROW();
	}
	PG_END_TRY();

	/* Remove self from dist db if no longer have data_nodes */
	if (data_node_get_node_name_list() == NIL)
		dist_util_remove_from_db();

	EventTriggerEndCompleteQuery();
	CommandCounterIncrement();
	CacheInvalidateRelcacheByRelid(ForeignServerRelationId);

	PG_RETURN_BOOL(true);
}

/*
 * Get server list, performing an ACL check on each of them in the process.
 */
List *
data_node_get_node_name_list_with_aclcheck(AclMode mode, bool fail_on_aclcheck)
{
	HeapTuple tuple;
	ScanKeyData scankey[1];
	SysScanDesc scandesc;
	Relation rel;
	ForeignDataWrapper *fdw = GetForeignDataWrapperByName(EXTENSION_FDW_NAME, false);
	List *nodes = NIL;

	rel = table_open(ForeignServerRelationId, AccessShareLock);

	ScanKeyInit(&scankey[0],
				Anum_pg_foreign_server_srvfdw,
				BTEqualStrategyNumber,
				F_OIDEQ,
				ObjectIdGetDatum(fdw->fdwid));

	scandesc = systable_beginscan(rel, InvalidOid, false, NULL, 1, scankey);

	while (HeapTupleIsValid(tuple = systable_getnext(scandesc)))
	{
		Form_pg_foreign_server form = (Form_pg_foreign_server) GETSTRUCT(tuple);
		ForeignServer *server;

		server =
			data_node_get_foreign_server(NameStr(form->srvname), mode, fail_on_aclcheck, false);

		if (server != NULL)
			nodes = lappend(nodes, pstrdup(NameStr(form->srvname)));
	}

	systable_endscan(scandesc);
	table_close(rel, AccessShareLock);

	return nodes;
}

bool
data_node_some_unavailable(void)
{
	/* Get a list of data nodes and check if one is unavailable */
	List *data_node_list = data_node_get_node_name_list_with_aclcheck(ACL_NO_CHECK, false);
	ListCell *lc;

	foreach (lc, data_node_list)
	{
		const char *node_name = lfirst(lc);
		const ForeignServer *server;

		server = data_node_get_foreign_server(node_name, ACL_NO_CHECK, false, false);
		if (!ts_data_node_is_available_by_server(server))
		{
			list_free(data_node_list);
			return true;
		}
	}
	list_free(data_node_list);
	return false;
}

/*
 * Get server list with optional ACL check.
 *
 * Returns:
 *
 * If nodearr is NULL, returns all system-configured data nodes that fulfill
 * the ACL check.
 *
 * If nodearr is non-NULL, returns all the data nodes in the specified array
 * subject to ACL checks.
 */
List *
data_node_get_filtered_node_name_list(ArrayType *nodearr, AclMode mode, bool fail_on_aclcheck)
{
	ArrayIterator it;
	Datum node_datum;
	bool isnull;
	List *nodes = NIL;

	if (NULL == nodearr)
		return data_node_get_node_name_list_with_aclcheck(mode, fail_on_aclcheck);

	it = array_create_iterator(nodearr, 0, NULL);

	while (array_iterate(it, &node_datum, &isnull))
	{
		if (!isnull)
		{
			const char *node_name = DatumGetCString(node_datum);
			ForeignServer *server =
				data_node_get_foreign_server(node_name, mode, fail_on_aclcheck, false);

			if (NULL != server)
				nodes = lappend(nodes, server->servername);
		}
	}

	array_free_iterator(it);

	return nodes;
}

List *
data_node_get_node_name_list(void)
{
	return data_node_get_node_name_list_with_aclcheck(ACL_NO_CHECK, false);
}

/*
 * Turn an array of data nodes into a list of names.
 *
 * The function will verify that all the servers in the list belong to the
 * TimescaleDB foreign data wrapper. Optionally, perform ACL check on each
 * data node's foreign server. Checks are skipped when specificing
 * ACL_NO_CHECK. If fail_on_aclcheck is false, then no errors will be thrown
 * on ACL check failures. Instead, data nodes that fail ACL checks will simply
 * be filtered.
 */
List *
data_node_array_to_node_name_list_with_aclcheck(ArrayType *nodearr, AclMode mode,
												bool fail_on_aclcheck)
{
	if (NULL == nodearr)
		return NIL;

	Assert(ARR_NDIM(nodearr) <= 1);

	return data_node_get_filtered_node_name_list(nodearr, mode, fail_on_aclcheck);
}

List *
data_node_array_to_node_name_list(ArrayType *nodearr)
{
	return data_node_array_to_node_name_list_with_aclcheck(nodearr, ACL_NO_CHECK, false);
}

Datum
data_node_ping(PG_FUNCTION_ARGS)
{
	const char *node_name = PG_ARGISNULL(0) ? NULL : PG_GETARG_CSTRING(0);
	Interval *timeout = PG_ARGISNULL(1) ? NULL : PG_GETARG_INTERVAL_P(1);
	TimestampTz endtime = TS_NO_TIMEOUT;

	/* Allow anyone to ping a data node. Otherwise the
	 * timescaledb_information.data_node view won't work for those users. */
	ForeignServer *server = data_node_get_foreign_server(node_name, ACL_NO_CHECK, false, false);
	bool success;

	Assert(NULL != server);

	/* Get endtime in microseconds */
	if (timeout)
		endtime = GetCurrentTimestamp() + ts_get_interval_period_approx(timeout);

	success = remote_connection_ping(server->servername, endtime);

	PG_RETURN_DATUM(BoolGetDatum(success));
}

List *
data_node_oids_to_node_name_list(List *data_node_oids, AclMode mode)
{
	List *node_names = NIL;
	ListCell *lc;
	ForeignServer *fs;

	foreach (lc, data_node_oids)
	{
		Oid foreign_server_oid = lfirst_oid(lc);
		fs = data_node_get_foreign_server_by_oid(foreign_server_oid, mode);
		node_names = lappend(node_names, pstrdup(fs->servername));
	}

	return node_names;
}

void
data_node_name_list_check_acl(List *data_node_names, AclMode mode)
{
	AclResult aclresult;
	Oid curuserid;
	ListCell *lc;

	if (data_node_names == NIL)
		return;

	curuserid = GetUserId();

	foreach (lc, data_node_names)
	{
		/* Validate the servers, but privilege check is optional */
		ForeignServer *server = GetForeignServerByName(lfirst(lc), false);

		if (mode != ACL_NO_CHECK)
		{
			/* Must have permissions on the server object */
			aclresult = pg_foreign_server_aclcheck(server->serverid, curuserid, mode);

			if (aclresult != ACLCHECK_OK)
				aclcheck_error(aclresult, OBJECT_FOREIGN_SERVER, server->servername);
		}
	}
}