377 lines
17 KiB
C
377 lines
17 KiB
C
/* -------------------------------------------------------------------------
|
|
*
|
|
* pgxcnode.h
|
|
*
|
|
* Utility functions to communicate to Datanodes and Coordinators
|
|
*
|
|
*
|
|
* Portions Copyright (c) 1996-2011, PostgreSQL Global Development Group ?
|
|
* Portions Copyright (c) 2010-2012 Postgres-XC Development Group
|
|
*
|
|
* src/include/pgxc/pgxcnode.h
|
|
*
|
|
* -------------------------------------------------------------------------
|
|
*/
|
|
|
|
#ifndef PGXCNODE_H
|
|
#define PGXCNODE_H
|
|
#include "postgres.h"
|
|
#include "knl/knl_variable.h"
|
|
#include "gtm/gtm_c.h"
|
|
#include "utils/timestamp.h"
|
|
#include "nodes/pg_list.h"
|
|
#include "utils/globalplancache.h"
|
|
#include "utils/snapshot.h"
|
|
#include "libcomm/libcomm.h"
|
|
#include "pgxc/poolmgr.h"
|
|
#include "getaddrinfo.h"
|
|
#include <unistd.h>
|
|
#include "knl/knl_variable.h"
|
|
#include <sys/time.h>
|
|
#include <sys/ioctl.h>
|
|
#include "access/transam.h"
|
|
#include "access/xact.h"
|
|
#include "catalog/pg_type.h"
|
|
#include "nodes/nodes.h"
|
|
#include "catalog/namespace.h"
|
|
#include "catalog/pg_namespace.h"
|
|
#include "catalog/pgxc_node.h"
|
|
#include "catalog/pg_collation.h"
|
|
#include "tcop/dest.h"
|
|
#include "utils/distribute_test.h"
|
|
#include "utils/elog.h"
|
|
#include "utils/guc.h"
|
|
#include "utils/memutils.h"
|
|
#include "utils/snapmgr.h"
|
|
#include "utils/syscache.h"
|
|
#include "utils/formatting.h"
|
|
#include "miscadmin.h"
|
|
#include "postmaster/postmaster.h"
|
|
#include "postmaster/twophasecleaner.h"
|
|
#include "libcomm/libcomm.h"
|
|
|
|
#ifdef HAVE_POLL_H
|
|
#include <poll.h>
|
|
#endif
|
|
#ifdef HAVE_SYS_POLL_H
|
|
#include <sys/poll.h>
|
|
#endif
|
|
|
|
#define NO_SOCKET -1
|
|
#define CMD_ID_MSG_LEN 8
|
|
#define PGXC_CANCEL_DELAY 15
|
|
|
|
#define ERROR_OCCURED true
|
|
#define NO_ERROR_OCCURED false
|
|
#define CN_SESSION_ID_MASK 0xffffffffffff
|
|
|
|
/* Message type code of bucket map and dn node index */
|
|
#define MSG_TYPE_PGXC_BUCKET_MAP 'G'
|
|
|
|
/* for logic connection sock is useless, type of gstock gives whether current logic connection is ok */
|
|
#define IS_VALID_CONNECTION(handle) \
|
|
((handle->is_logic_conn) ? (handle->gsock.type != GSOCK_INVALID) : (handle->sock != NO_SOCKET))
|
|
|
|
#define DDL_PBE_VERSION_NUM 92067 /* Version control for DDL PBE */
|
|
|
|
struct StreamNetCtl;
|
|
|
|
/* Helper structure to access Datanode from Session */
|
|
typedef enum {
|
|
DN_CONNECTION_STATE_IDLE, /* idle, ready for query */
|
|
DN_CONNECTION_STATE_QUERY, /* query is sent, response expected */
|
|
DN_CONNECTION_STATE_ERROR_FATAL, /* fatal error */
|
|
DN_CONNECTION_STATE_COPY_IN,
|
|
DN_CONNECTION_STATE_COPY_OUT
|
|
} DNConnectionState;
|
|
|
|
typedef enum { HANDLE_IDLE, HANDLE_ERROR, HANDLE_DEFAULT } PGXCNode_HandleRequested;
|
|
|
|
/*
|
|
* Enumeration for two purposes
|
|
* 1. To indicate to the HandleCommandComplete function whether response checking is required or not
|
|
* 2. To enable HandleCommandComplete function to indicate whether the response was a ROLLBACK or not
|
|
* Response checking is required in case of PREPARE TRANSACTION and should not be done for the rest
|
|
* of the cases for performance reasons, hence we have an option to ignore response checking.
|
|
* The problem with PREPARE TRANSACTION is that it can result in a ROLLBACK response
|
|
* yet Coordinator would think it got done on all nodes.
|
|
* If we ignore ROLLBACK response then we would try to COMMIT a transaction that
|
|
* never got prepared, which in an incorrect behavior.
|
|
*/
|
|
typedef enum {
|
|
RESP_ROLLBACK_IGNORE, /* Ignore response checking */
|
|
RESP_ROLLBACK_CHECK, /* Check whether response was ROLLBACK */
|
|
RESP_ROLLBACK_RECEIVED, /* Response is ROLLBACK */
|
|
RESP_ROLLBACK_NOT_RECEIVED /* Response is NOT ROLLBACK */
|
|
} RESP_ROLLBACK;
|
|
|
|
/* To identify the different uses of CSN */
|
|
typedef enum {
|
|
FETCH_CSN, /* A flag to fecth csn from other nodes */
|
|
GLOBAL_CSN_MIN /* The global min csn between all nodes */
|
|
} CsnType;
|
|
|
|
#define DN_CONNECTION_STATE_ERROR(dnconn) \
|
|
((dnconn)->state == DN_CONNECTION_STATE_ERROR_FATAL || (dnconn)->transaction_status == 'E')
|
|
|
|
#define HAS_MESSAGE_BUFFERED(conn) \
|
|
((conn)->inCursor + 4 < (conn)->inEnd && \
|
|
(conn)->inCursor + ntohl(*((uint32_t*)((conn)->inBuffer + (conn)->inCursor + 1))) < (conn)->inEnd)
|
|
|
|
/*
|
|
* Represents a DataRow message received from a remote node.
|
|
* Contains originating node number and message body in DataRow format without
|
|
* message code and length. Length is separate field
|
|
*/
|
|
typedef struct RemoteDataRowData {
|
|
char* msg; /* last data row message */
|
|
int msglen; /* length of the data row message */
|
|
int msgnode; /* node number of the data row message */
|
|
} RemoteDataRowData;
|
|
typedef RemoteDataRowData* RemoteDataRow;
|
|
|
|
struct pgxc_node_handle {
|
|
Oid nodeoid;
|
|
|
|
/* fd of the connection */
|
|
int sock;
|
|
|
|
/* tcp connection */
|
|
int tcpCtlPort;
|
|
int listenPort;
|
|
/* logic connection socket between cn and dn */
|
|
gsocket gsock;
|
|
|
|
char transaction_status;
|
|
DNConnectionState state;
|
|
struct RemoteQueryState* combiner;
|
|
struct StreamState* stream;
|
|
#ifdef DN_CONNECTION_DEBUG
|
|
bool have_row_desc;
|
|
#endif
|
|
char* error;
|
|
char* hint;
|
|
/* Output buffer */
|
|
char* outBuffer;
|
|
size_t outSize;
|
|
size_t outEnd;
|
|
uint64 outNum;
|
|
/* Input buffer */
|
|
char* inBuffer;
|
|
size_t inSize;
|
|
size_t inStart;
|
|
size_t inEnd;
|
|
size_t inCursor;
|
|
|
|
/*
|
|
* Have a variable to enable/disable response checking and
|
|
* if enable then read the result of response checking
|
|
*
|
|
* For details see comments of RESP_ROLLBACK
|
|
*/
|
|
RESP_ROLLBACK ck_resp_rollback;
|
|
|
|
char* remoteNodeName;
|
|
int nodeIdx; /* datanode index, like t_thrd.postmaster_cxt.PGXCNodeId */
|
|
knl_virtual_role remote_node_type;
|
|
PoolConnInfo connInfo;
|
|
/* flag to identify logic connection between cn and dn */
|
|
bool is_logic_conn;
|
|
NODE_CONNECTION* pg_conn;
|
|
MessageCommLog *msgLog;
|
|
TransactionId remote_top_txid;
|
|
};
|
|
typedef struct pgxc_node_handle PGXCNodeHandle;
|
|
|
|
/*
|
|
* Because the number of node handles is identical to the number of nodes,
|
|
* so here we combine both CN and DN handles together for better maitanences
|
|
*/
|
|
typedef struct PGXCNodeHandleGroup {
|
|
PGXCNodeHandle* nodeHandles;
|
|
uint32 nodeHandleNums;
|
|
|
|
/* dn and cn slots often process together, so need offset to indicate real postion */
|
|
uint32 offset;
|
|
} PGXCNodeHandleGroup;
|
|
|
|
typedef struct PGXCNodeNetCtlLayer {
|
|
int conn_num; /* size of array */
|
|
struct pollfd* ufds; /* physic poll fds. */
|
|
int* datamarks; /* producer number triggers poll. */
|
|
int* poll2conn; /* poll idx to connection idx. */
|
|
gsocket* gs_sock; /* logic poll gs_sock. */
|
|
} PGXCNodeNetCtlLayer;
|
|
|
|
/* Structure used to get all the handles involved in a transaction */
|
|
typedef struct PGXCNodeAllHandles {
|
|
PGXCNodeHandle* primary_handle; /* Primary connection to PGXC node */
|
|
int dn_conn_count; /* number of Datanode Handles including primary handle */
|
|
PGXCNodeHandle** datanode_handles; /* an array of Datanode handles */
|
|
List* dummy_datanode_handles; /* only valid for replicate table dml */
|
|
int co_conn_count; /* number of Coordinator handles */
|
|
PGXCNodeHandle** coord_handles; /* an array of Coordinator handles */
|
|
} PGXCNodeAllHandles;
|
|
|
|
extern unsigned long LIBCOMM_BUFFER_SIZE;
|
|
extern void InitMultinodeExecutor(bool is_force);
|
|
|
|
/* Open/close connection routines (invoked from Pool Manager) */
|
|
extern char *PGXCNodeConnStr(const char *host, int port, const char *dbname, const char *user,
|
|
const char *pgoptions, const char *remote_type, int proto_type, const char *remote_nodename);
|
|
extern NODE_CONNECTION* PGXCNodeConnect(char* connstr);
|
|
extern int PGXCNodeSendSetQuery(NODE_CONNECTION* conn, const char* sql_command);
|
|
extern int PGXCNodeSetQueryGetResult(NODE_CONNECTION* conn);
|
|
extern int PGXCNodeSendSetQueryPoolerStatelessReuse(NODE_CONNECTION* conn, const char* sql_command);
|
|
extern void PGXCNodeClose(NODE_CONNECTION* conn);
|
|
extern int PGXCNodeConnected(NODE_CONNECTION* conn);
|
|
extern int PGXCNodeConnClean(NODE_CONNECTION* conn);
|
|
extern void PGXCNodeCleanAndRelease(int code, Datum arg);
|
|
extern void PGXCConnClean(int code, Datum arg);
|
|
|
|
/* Look at information cached in node handles */
|
|
extern int PGXCNodeGetNodeId(Oid nodeoid, char node_type);
|
|
extern int PGXCNodeDRGetNodeId(Oid nodeoid);
|
|
extern Oid PGXCNodeGetNodeOid(int nodeid, char node_type);
|
|
extern int PGXCNodeGetNodeIdFromName(const char* node_name, char node_type);
|
|
extern char* PGXCNodeGetNodeNameFromId(int nodeid, char node_type);
|
|
extern int LibcommPacketSend(NODE_CONNECTION* conn, char pack_type, const void* buf, size_t buf_len);
|
|
extern bool LibcommCancelOrStop(NODE_CONNECTION* conn, char* errbuf, int errbufsize, int timeout, uint32 request_code);
|
|
extern bool LibcommStopQuery(NODE_CONNECTION* conn);
|
|
extern int LibcommSendSome(NODE_CONNECTION* conn, int len);
|
|
extern int LibcommPutMsgEnd(NODE_CONNECTION* conn);
|
|
extern int LibcommFlush(NODE_CONNECTION* conn);
|
|
extern int LibcommSendQuery(NODE_CONNECTION* conn, const char* query);
|
|
extern int LibcommReadData(NODE_CONNECTION* conn);
|
|
extern PGXCNodeAllHandles* get_handles(
|
|
List* datanodelist, List* coordlist, bool is_query_coord_only, List* dummydatanodelist = NIL);
|
|
extern void pfree_pgxc_all_handles(PGXCNodeAllHandles* handles);
|
|
extern void release_pgxc_handles(PGXCNodeAllHandles* pgxc_handles);
|
|
extern void release_handles(void);
|
|
extern void destroy_handles(void);
|
|
extern void reset_handles_at_abort(void);
|
|
extern void cancel_query(void);
|
|
extern void cancel_query_without_read(void);
|
|
extern void stop_query(void);
|
|
extern void clear_all_data(void);
|
|
extern int get_transaction_nodes(
|
|
PGXCNodeHandle** connections, char client_conn_type, PGXCNode_HandleRequested type_requested);
|
|
extern char* collect_pgxcnode_names(
|
|
char* nodestring, int conn_count, PGXCNodeHandle** connections, char client_conn_type);
|
|
extern char* collect_localnode_name(char* nodestring);
|
|
extern int get_active_nodes(PGXCNodeHandle** connections);
|
|
extern void ensure_in_buffer_capacity(size_t bytes_needed, PGXCNodeHandle* handle);
|
|
extern void ensure_out_buffer_capacity(size_t bytes_needed, PGXCNodeHandle* handle);
|
|
extern int pgxc_node_send_threadid(PGXCNodeHandle* handle, uint32 threadid);
|
|
extern int pgxc_node_send_queryid(PGXCNodeHandle* handle, uint64 queryid);
|
|
extern int pgxc_node_send_unique_sql_id(PGXCNodeHandle* handle);
|
|
extern int pgxc_node_send_global_session_id(PGXCNodeHandle *handle);
|
|
extern int pgxc_node_send_query(PGXCNodeHandle* handle, const char* query, bool isPush = false,
|
|
bool isCreateSchemaPush = false, bool trigger_ship = false, bool check_gtm_mode = false,
|
|
const char *compressPlan = NULL, int cLen = 0);
|
|
extern void pgxc_node_send_gtm_mode(PGXCNodeHandle* handle);
|
|
extern int pgxc_node_send_plan_with_params(PGXCNodeHandle* handle, const char* query, short num_params,
|
|
Oid* param_types, int paramlen, const char* params, int fetch_size);
|
|
extern int pgxc_node_send_describe(PGXCNodeHandle* handle, bool is_statement, const char* name);
|
|
#ifndef ENABLE_MULTIPLE_NODES
|
|
extern int pgxc_node_send_execute(PGXCNodeHandle* handle, const char* portal, int fetch);
|
|
#endif
|
|
extern int pgxc_node_send_close(PGXCNodeHandle* handle, bool is_statement, const char* name);
|
|
extern int pgxc_node_send_sync(PGXCNodeHandle* handle);
|
|
extern int pgxc_node_send_bind(
|
|
PGXCNodeHandle* handle, const char* portal, const char* statement, int paramlen, const char* params);
|
|
extern int pgxc_node_send_parse(
|
|
PGXCNodeHandle* handle, const char* statement, const char* query, short num_params, Oid* param_types);
|
|
#ifndef ENABLE_MULTIPLE_NODES
|
|
extern int pgxc_node_send_query_extended(PGXCNodeHandle* handle, const char* query, const char* statement,
|
|
const char* portal, int num_params, Oid* param_types, int paramlen, const char* params, bool send_describe,
|
|
int fetch_size);
|
|
#endif
|
|
extern int pgxc_node_send_gxid(PGXCNodeHandle* handle, GlobalTransactionId gxid, bool isforcheck);
|
|
extern int pgxc_node_send_commit_csn(PGXCNodeHandle* handle, uint64 commit_csn);
|
|
extern int pgxc_node_send_csn(PGXCNodeHandle* handle, uint64 commit_csn, CsnType csn_type);
|
|
extern int pgxc_node_notify_commit(PGXCNodeHandle* handle);
|
|
extern int pgxc_node_send_cmd_id(PGXCNodeHandle* handle, CommandId cid);
|
|
extern int pgxc_node_send_wlm_cgroup(PGXCNodeHandle* handle);
|
|
extern int pgxc_node_dywlm_send_params_for_jobs(PGXCNodeHandle* handle, int tag, const char* keystr);
|
|
extern int pgxc_node_dywlm_send_params(PGXCNodeHandle* handle, const void* info);
|
|
extern int pgxc_node_dywlm_send_record(PGXCNodeHandle* handle, int tag, const char* infostr);
|
|
extern int pgxc_node_send_pgfdw(PGXCNodeHandle* handle, int tag, const char* keystr, int len);
|
|
extern int pgxc_node_send_snapshot(PGXCNodeHandle* handle, Snapshot snapshot, int max_push_sqls = 1);
|
|
extern int pgxc_node_send_ddl_params(PGXCNodeHandle* handle);
|
|
extern int pgxc_node_send_timestamp(
|
|
PGXCNodeHandle* handle, TimestampTz gtmstart_timestamp, TimestampTz stmtsys_timestamp);
|
|
extern int pgxc_node_send_pushschema(PGXCNodeHandle* handle, bool* isPush, bool *isCreateSchemaPush);
|
|
extern int pgxc_node_send_popschema(PGXCNodeHandle* handle, bool isPush);
|
|
extern int pgxc_node_is_data_enqueued(PGXCNodeHandle* conn);
|
|
extern int send_some(PGXCNodeHandle* handle, int len);
|
|
extern int pgxc_node_flush(PGXCNodeHandle* handle);
|
|
extern bool pgxc_node_flush_read(PGXCNodeHandle* handle);
|
|
extern int pgxc_all_handles_send_gxid(PGXCNodeAllHandles* pgxc_handles, GlobalTransactionId gxid, bool stop_at_error);
|
|
extern int pgxc_all_handles_send_query(PGXCNodeAllHandles* pgxc_handles, const char* buffer, bool stop_at_error);
|
|
extern void add_error_message(PGXCNodeHandle* handle, const char *message, ...);
|
|
extern char get_message(PGXCNodeHandle *conn, int *len, char **msg);
|
|
extern Datum pgxc_execute_on_nodes(int numnodes, Oid* nodelist, char* query);
|
|
extern bool pgxc_node_receive(const int conn_count, PGXCNodeHandle** connections,
|
|
struct timeval* timeout, bool ignoreTimeoutWarning = false);
|
|
extern bool datanode_receive_from_physic_conn(
|
|
const int conn_count, PGXCNodeHandle** connections, struct timeval* timeout);
|
|
extern bool datanode_receive_from_logic_conn(
|
|
const int conn_count, PGXCNodeHandle** connections, StreamNetCtl* ctl, int time_out);
|
|
extern bool pgxc_node_validate(PGXCNodeHandle *conn);
|
|
extern int pgxc_node_read_data(PGXCNodeHandle* conn, bool close_if_error, bool StreamConnection = false);
|
|
extern int pgxc_node_read_data_from_logic_conn(PGXCNodeHandle* conn, bool close_if_error);
|
|
extern void ResetHandleOutBuffer(PGXCNodeHandle* handle);
|
|
extern void connect_server(
|
|
const char* conn_str, PGXCNodeHandle** handle, const char* host, int port, const char* log_conn_str);
|
|
extern void release_conn_to_compute_pool();
|
|
extern void release_pgfdw_conn();
|
|
extern int pgxc_node_send_userpl(PGXCNodeHandle* handle, int userpl);
|
|
extern int pgxc_node_send_userpl(PGXCNodeHandle* handle, int64 userpl);
|
|
extern int pgxc_node_send_cpconf(PGXCNodeHandle* handle, char* data);
|
|
extern int pgxc_send_bucket_map(PGXCNodeHandle* handle, uint2* bucketMap, int dnNodeIdx, int bucketCnt);
|
|
extern bool light_node_receive(PGXCNodeHandle* handle);
|
|
extern bool light_node_receive_from_logic_conn(PGXCNodeHandle* handle);
|
|
extern int pgxc_node_pgstat_send(PGXCNodeHandle* handle, char tag);
|
|
extern void pgxc_node_free(PGXCNodeHandle* handle);
|
|
extern void pgxc_node_init(PGXCNodeHandle* handle, int sock);
|
|
extern void pgxc_node_free_def(PoolConnDef* result);
|
|
extern void pgxc_palloc_net_ctl(int conn_num);
|
|
extern int ts_get_tag_colnum(Oid relid, bool need_index = false, List** idx_list = NULL);
|
|
extern void init_pgxc_handle(PGXCNodeHandle* pgxc_handle);
|
|
|
|
#ifdef ENABLE_MULTIPLE_NODES
|
|
extern GlobalNodeDefinition *global_node_definition;
|
|
extern bool IsAutoVacuumWorkerProcess();
|
|
extern char *format_type_with_nspname(Oid type_oid);
|
|
extern void pgxc_node_all_free(void);
|
|
extern int pgxc_node_send_execute(PGXCNodeHandle* handle, const char* portal, int fetch, bool trigger_ship = false);
|
|
extern int pgxc_node_send_query_extended(PGXCNodeHandle* handle, const char* query, const char* statement,
|
|
const char* portal, int num_params, Oid* param_types, int paramlen, const char* params, bool send_describe,
|
|
int fetch_size, bool trigger_ship = false);
|
|
extern void FreePgxcNodehandles(PGXCNodeHandleGroup* handleGroup);
|
|
extern void FreePgxcNodehandlesWithStatus(PGXCNodeHandleGroup* handleGroup, char *statusArray, bool *hasError);
|
|
extern void LibcommFinish(NODE_CONNECTION *conn);
|
|
extern int LibcommWaitpoll(NODE_CONNECTION *conn);
|
|
extern int get_int(PGXCNodeHandle * conn, size_t len, int *out);
|
|
extern int get_char(PGXCNodeHandle * conn, char *out);
|
|
extern PoolHandle * GetPoolAgent();
|
|
extern void destroy_slots(List *slots);
|
|
extern void init_pgxc_handle(PGXCNodeHandle *pgxc_handle);
|
|
extern char* get_pgxc_nodename(Oid nodeid, NameData* namedata);
|
|
extern List* get_dn_handles(List *datanodelist, List *dummynodelist, PGXCNodeAllHandles* result);
|
|
extern List* get_cn_handles(List *coordlist, PGXCNodeAllHandles* result);
|
|
extern void init_handles(List* dn_allocate, List* co_allocate, PoolConnDef* pfds);
|
|
extern int PollInterrupts(struct pollfd *ufds, int nfds, int timeout);
|
|
extern int find_name_by_params(Oid *param_types, char **paramTypes, short num_params);
|
|
extern void add_params_name_to_send(PGXCNodeHandle *handle, char **paramTypes, short num_params);
|
|
extern int LibcommSendQueryPoolerStatelessReuse(NODE_CONNECTION *conn, const char *query);
|
|
#endif
|
|
|
|
#ifdef ENABLE_UT
|
|
extern char* getNodenameByIndex(int index);
|
|
extern int light_node_read_data(PGXCNodeHandle* conn);
|
|
#endif /* USE_UT */
|
|
|
|
#endif /* PGXCNODE_H */
|