自学教程：C++ GetTransactionSnapshot函数代码示例

/* * We keep some resources across transactions, so we attach everything to a * long-lived ResourceOwner, which prevents the below commit from thinking that * there are reference leaks */static voidstart_executor(QueryDesc *queryDesc, MemoryContext context, ResourceOwner owner){	MemoryContext old;	ResourceOwner save;	StartTransactionCommand();	old = MemoryContextSwitchTo(context);	save = CurrentResourceOwner;	CurrentResourceOwner = owner;	queryDesc->snapshot = GetTransactionSnapshot();	queryDesc->snapshot->copied = true;	RegisterSnapshotOnOwner(queryDesc->snapshot, owner);	ExecutorStart(queryDesc, 0);	queryDesc->snapshot->active_count++;	UnregisterSnapshotFromOwner(queryDesc->snapshot, owner);	UnregisterSnapshotFromOwner(queryDesc->estate->es_snapshot, owner);	CurrentResourceOwner = TopTransactionResourceOwner;	MemoryContextSwitchTo(old);	CommitTransactionCommand();	CurrentResourceOwner = save;}

/* * get_all_brokers * * Return a list of all brokers in pipeline_kafka_brokers */static List *get_all_brokers(void){	HeapTuple tup = NULL;	HeapScanDesc scan;	Relation brokers = open_pipeline_kafka_brokers();	TupleTableSlot *slot = MakeSingleTupleTableSlot(RelationGetDescr(brokers));	List *result = NIL;	scan = heap_beginscan(brokers, GetTransactionSnapshot(), 0, NULL);	while ((tup = heap_getnext(scan, ForwardScanDirection)) != NULL)	{		char *host;		Datum d;		bool isnull;		ExecStoreTuple(tup, slot, InvalidBuffer, false);		d = slot_getattr(slot, BROKER_ATTR_HOST, &isnull);		host = TextDatumGetCString(d);		result = lappend(result, host);	}	ExecDropSingleTupleTableSlot(slot);	heap_endscan(scan);	heap_close(brokers, NoLock);	return result;}

/* * Returns a count of the number of non-template databases from the catalog. */int get_database_count(void) {  int retval, processed;	StringInfoData buf;	SPITupleTable *coltuptable;	int database_count = 0;    SetCurrentStatementStartTimestamp();  StartTransactionCommand();  SPI_connect();  PushActiveSnapshot(GetTransactionSnapshot());    initStringInfo(&buf);  appendStringInfo(&buf, "SELECT count(*) FROM pg_database WHERE datname NOT IN ('template0', 'template1') AND datallowconn IS TRUE;");    retval = SPI_execute(buf.data, false, 0);  if (retval != SPI_OK_SELECT) {    elog(FATAL, "Database information collection failed");    // FAIL RETURN 1  }    processed = SPI_processed;    if (processed > 0) {    coltuptable = SPI_tuptable;      database_count = atoi(SPI_getvalue(coltuptable->vals[0], coltuptable->tupdesc, 1));  }    SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();     return database_count;}

/* * GetLatestSnapshot *		Get a snapshot that is up-to-date as of the current instant, *		even if we are executing in transaction-snapshot mode. */SnapshotGetLatestSnapshot(void){	/*	 * We might be able to relax this, but nothing that could otherwise work	 * needs it.	 */	if (IsInParallelMode())		elog(ERROR,			 "cannot update SecondarySnapshot during a parallel operation");	/*	 * So far there are no cases requiring support for GetLatestSnapshot()	 * during logical decoding, but it wouldn't be hard to add if required.	 */	Assert(!HistoricSnapshotActive());	/* If first call in transaction, go ahead and set the xact snapshot */	if (!FirstSnapshotSet)		return GetTransactionSnapshot();	SecondarySnapshot = GetSnapshotData(&SecondarySnapshotData);	return SecondarySnapshot;}

/* * Main worker routine. Accepts dsm_handle as an argument */static voidbg_worker_main(Datum main_arg){	PartitionArgs  *args;	dsm_handle		handle = DatumGetInt32(main_arg);	/* Create resource owner */	CurrentResourceOwner = ResourceOwnerCreate(NULL, "CreatePartitionsWorker");	/* Attach to dynamic shared memory */	if (!handle)	{		ereport(WARNING,                (errmsg("pg_pathman worker: invalid dsm_handle")));	}	segment = dsm_attach(handle);	args = dsm_segment_address(segment);	/* Establish connection and start transaction */	BackgroundWorkerInitializeConnectionByOid(args->dbid, InvalidOid);	StartTransactionCommand();	SPI_connect();	PushActiveSnapshot(GetTransactionSnapshot());	/* Create partitions */	args->result = create_partitions(args->relid, PATHMAN_GET_DATUM(args->value, args->by_val), args->value_type, &args->crashed);	/* Cleanup */	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();	dsm_detach(segment);}

voidBeginScanAppendOnlyRelation(ScanState *scanState){    Snapshot appendOnlyMetaDataSnapshot;    Assert(IsA(scanState, TableScanState) ||           IsA(scanState, DynamicTableScanState));    AppendOnlyScanState *node = (AppendOnlyScanState *)scanState;    Assert(node->ss.scan_state == SCAN_INIT ||           node->ss.scan_state == SCAN_DONE);    Assert(node->aos_ScanDesc == NULL);    appendOnlyMetaDataSnapshot = node->ss.ps.state->es_snapshot;    if (appendOnlyMetaDataSnapshot == SnapshotAny)    {        /*         * the append-only meta data should never be fetched with         * SnapshotAny as bogus results are returned.         */        appendOnlyMetaDataSnapshot = GetTransactionSnapshot();    }    node->aos_ScanDesc = appendonly_beginscan(                             node->ss.ss_currentRelation,                             node->ss.ps.state->es_snapshot,                             appendOnlyMetaDataSnapshot,                             0, NULL);    node->ss.scan_state = SCAN_SCAN;}

/* * get_database_oids * * Returns a list of all database OIDs found in pg_database. */static List *get_database_list(void){	List *dbs = NIL;	Relation rel;	HeapScanDesc scan;	HeapTuple tup;	MemoryContext resultcxt;	/* This is the context that we will allocate our output data in */	resultcxt = CurrentMemoryContext;	/*	 * Start a transaction so we can access pg_database, and get a snapshot.	 * We don't have a use for the snapshot itself, but we're interested in	 * the secondary effect that it sets RecentGlobalXmin.  (This is critical	 * for anything that reads heap pages, because HOT may decide to prune	 * them even if the process doesn't attempt to modify any tuples.)	 */	StartTransactionCommand();	(void) GetTransactionSnapshot();	/* We take a AccessExclusiveLock so we don't conflict with any DATABASE commands */	rel = heap_open(DatabaseRelationId, AccessExclusiveLock);	scan = heap_beginscan_catalog(rel, 0, NULL);	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))	{		MemoryContext oldcxt;		Form_pg_database pgdatabase = (Form_pg_database) GETSTRUCT(tup);		DatabaseEntry *db_entry;		/* Ignore template databases or ones that don't allow connections. */		if (pgdatabase->datistemplate || !pgdatabase->datallowconn)			continue;		/*		 * Allocate our results in the caller's context, not the		 * transaction's. We do this inside the loop, and restore the original		 * context at the end, so that leaky things like heap_getnext() are		 * not called in a potentially long-lived context.		 */		oldcxt = MemoryContextSwitchTo(resultcxt);		db_entry = palloc0(sizeof(DatabaseEntry));		db_entry->oid = HeapTupleGetOid(tup);		StrNCpy(NameStr(db_entry->name), NameStr(pgdatabase->datname), NAMEDATALEN);		dbs = lappend(dbs, db_entry);		MemoryContextSwitchTo(oldcxt);	}	heap_endscan(scan);	heap_close(rel, AccessExclusiveLock);	CommitTransactionCommand();	return dbs;}

voidSetEStateSnapshot(EState *estate, ResourceOwner owner){	estate->es_snapshot = GetTransactionSnapshot();	estate->es_snapshot->active_count++;	estate->es_snapshot->copied = true;	PushActiveSnapshot(estate->es_snapshot);}

/* * Load the list of subscriptions. * * Only the fields interesting for worker start/stop functions are filled for * each subscription. */static List *get_subscription_list(void){	List	   *res = NIL;	Relation	rel;	TableScanDesc scan;	HeapTuple	tup;	MemoryContext resultcxt;	/* This is the context that we will allocate our output data in */	resultcxt = CurrentMemoryContext;	/*	 * Start a transaction so we can access pg_database, and get a snapshot.	 * We don't have a use for the snapshot itself, but we're interested in	 * the secondary effect that it sets RecentGlobalXmin.  (This is critical	 * for anything that reads heap pages, because HOT may decide to prune	 * them even if the process doesn't attempt to modify any tuples.)	 */	StartTransactionCommand();	(void) GetTransactionSnapshot();	rel = table_open(SubscriptionRelationId, AccessShareLock);	scan = table_beginscan_catalog(rel, 0, NULL);	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))	{		Form_pg_subscription subform = (Form_pg_subscription) GETSTRUCT(tup);		Subscription *sub;		MemoryContext oldcxt;		/*		 * Allocate our results in the caller's context, not the		 * transaction's. We do this inside the loop, and restore the original		 * context at the end, so that leaky things like heap_getnext() are		 * not called in a potentially long-lived context.		 */		oldcxt = MemoryContextSwitchTo(resultcxt);		sub = (Subscription *) palloc0(sizeof(Subscription));		sub->oid = subform->oid;		sub->dbid = subform->subdbid;		sub->owner = subform->subowner;		sub->enabled = subform->subenabled;		sub->name = pstrdup(NameStr(subform->subname));		/* We don't fill fields we are not interested in. */		res = lappend(res, sub);		MemoryContextSwitchTo(oldcxt);	}	table_endscan(scan);	table_close(rel, AccessShareLock);	CommitTransactionCommand();	return res;}

/* * Initialize workspace for a worker process: create the schema if it doesn't * already exist. */static voidinitialize_worker_spi(worktable *table){	int			ret;	int			ntup;	bool		isnull;	StringInfoData buf;	SetCurrentStatementStartTimestamp();	StartTransactionCommand();	SPI_connect();	PushActiveSnapshot(GetTransactionSnapshot());	pgstat_report_activity(STATE_RUNNING, "initializing spi_worker schema");	/* XXX could we use CREATE SCHEMA IF NOT EXISTS? */	initStringInfo(&buf);	appendStringInfo(&buf, "select count(*) from pg_namespace where nspname = '%s'",					 table->schema);	ret = SPI_execute(buf.data, true, 0);	if (ret != SPI_OK_SELECT)		elog(FATAL, "SPI_execute failed: error code %d", ret);	if (SPI_processed != 1)		elog(FATAL, "not a singleton result");	ntup = DatumGetInt64(SPI_getbinval(SPI_tuptable->vals[0],									   SPI_tuptable->tupdesc,									   1, &isnull));	if (isnull)		elog(FATAL, "null result");	if (ntup == 0)	{		resetStringInfo(&buf);		appendStringInfo(&buf,						 "CREATE SCHEMA /"%s/" "						 "CREATE TABLE /"%s/" ("			   "		type text CHECK (type IN ('total', 'delta')), "						 "		value	integer)"				  "CREATE UNIQUE INDEX /"%s_unique_total/" ON /"%s/" (type) "						 "WHERE type = 'total'",					   table->schema, table->name, table->name, table->name);		/* set statement start time */		SetCurrentStatementStartTimestamp();		ret = SPI_execute(buf.data, false, 0);		if (ret != SPI_OK_UTILITY)			elog(FATAL, "failed to create my schema");	}	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();	pgstat_report_activity(STATE_IDLE, NULL);}

/* *  Ensure that the environment is sane.   *    This involves checking the Postgresql version, and if in network mode *      also establishing a connection to a receiver.*/int ensure_valid_environment(void) {  StringInfoData buf;  int     retval;	char* pgversion;	  SPITupleTable *coltuptable;   SetCurrentStatementStartTimestamp();  StartTransactionCommand();  SPI_connect();  PushActiveSnapshot(GetTransactionSnapshot());    /* Ensure compatible version */  pgstat_report_activity(STATE_RUNNING, "verifying compatible postgres version");  initStringInfo(&buf);  appendStringInfo(&buf,     "select version();"  );  retval = SPI_execute(buf.data, false, 0);  if (retval != SPI_OK_SELECT) {    elog(FATAL, "Unable to query postgres version %d", retval);    SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();  	return 1;    }  	coltuptable = SPI_tuptable;	pgversion = SPI_getvalue(coltuptable->vals[0], coltuptable->tupdesc, 1);	  if(strstr(pgversion, "PostgreSQL 9.3") == NULL) {    elog(FATAL, "Unsupported Postgresql version");    SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();  	return 1;	}  	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();    /*   * Attempt to establish a connection if the output mode is network.   */  if (strcmp(output_mode, "network") == 0) {		retval = establish_connection();		if (retval == 2) {			elog(LOG, "Error : Failed to connect to antenna please check domain is available from host.");		}	}		//TODO verify logging directory is accessible when csv mode.	  elog(LOG, "Pgsampler Initialized");  return 0;}

/* * Handle INSERT message. */static voidapply_handle_insert(StringInfo s){	LogicalRepRelMapEntry *rel;	LogicalRepTupleData newtup;	LogicalRepRelId relid;	EState	   *estate;	TupleTableSlot *remoteslot;	MemoryContext oldctx;	ensure_transaction();	relid = logicalrep_read_insert(s, &newtup);	rel = logicalrep_rel_open(relid, RowExclusiveLock);	if (!should_apply_changes_for_rel(rel))	{		/*		 * The relation can't become interesting in the middle of the		 * transaction so it's safe to unlock it.		 */		logicalrep_rel_close(rel, RowExclusiveLock);		return;	}	/* Initialize the executor state. */	estate = create_estate_for_relation(rel);	remoteslot = ExecInitExtraTupleSlot(estate,										RelationGetDescr(rel->localrel));	/* Input functions may need an active snapshot, so get one */	PushActiveSnapshot(GetTransactionSnapshot());	/* Process and store remote tuple in the slot */	oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));	slot_store_cstrings(remoteslot, rel, newtup.values);	slot_fill_defaults(rel, estate, remoteslot);	MemoryContextSwitchTo(oldctx);	ExecOpenIndices(estate->es_result_relation_info, false);	/* Do the insert. */	ExecSimpleRelationInsert(estate, remoteslot);	/* Cleanup. */	ExecCloseIndices(estate->es_result_relation_info);	PopActiveSnapshot();	/* Handle queued AFTER triggers. */	AfterTriggerEndQuery(estate);	ExecResetTupleTable(estate->es_tupleTable, false);	FreeExecutorState(estate);	logicalrep_rel_close(rel, NoLock);	CommandCounterIncrement();}

static voidset_snapshot(EState *estate, ResourceOwner owner){	estate->es_snapshot = GetTransactionSnapshot();	estate->es_snapshot->active_count++;	estate->es_snapshot->copied = true;	RegisterSnapshotOnOwner(estate->es_snapshot, owner);	PushActiveSnapshot(estate->es_snapshot);}

/* * GetLatestSnapshot *		Get a snapshot that is up-to-date as of the current instant, *		even if we are executing in transaction-snapshot mode. */SnapshotGetLatestSnapshot(void){	/* If first call in transaction, go ahead and set the xact snapshot */	if (!FirstSnapshotSet)		return GetTransactionSnapshot();	SecondarySnapshot = GetSnapshotData(&SecondarySnapshotData);	return SecondarySnapshot;}

/* * load_consumer_offsets * * Load all offsets for all of this consumer's partitions */static voidload_consumer_offsets(KafkaConsumer *consumer, struct rd_kafka_metadata_topic *meta, int64_t offset){	MemoryContext old;	ScanKeyData skey[1];	HeapTuple tup = NULL;	HeapScanDesc scan;	Relation offsets = open_pipeline_kafka_offsets();	TupleTableSlot *slot = MakeSingleTupleTableSlot(RelationGetDescr(offsets));	int i;	ScanKeyInit(&skey[0], OFFSETS_ATTR_CONSUMER, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(consumer->id));	scan = heap_beginscan(offsets, GetTransactionSnapshot(), 1, skey);	old = MemoryContextSwitchTo(CacheMemoryContext);	consumer->offsets = palloc0(meta->partition_cnt * sizeof(int64_t));	MemoryContextSwitchTo(old);	/* by default, begin consuming from the end of a stream */	for (i = 0; i < meta->partition_cnt; i++)		consumer->offsets[i] = offset;	consumer->num_partitions = meta->partition_cnt;	while ((tup = heap_getnext(scan, ForwardScanDirection)) != NULL)	{		Datum d;		bool isnull;		int partition;		ExecStoreTuple(tup, slot, InvalidBuffer, false);		d = slot_getattr(slot, OFFSETS_ATTR_PARTITION, &isnull);		partition = DatumGetInt32(d);		if(partition > consumer->num_partitions)			elog(ERROR, "invalid partition id: %d", partition);		if (offset == RD_KAFKA_OFFSET_NULL)		{			d = slot_getattr(slot, OFFSETS_ATTR_OFFSET, &isnull);			if (isnull)				offset = RD_KAFKA_OFFSET_END;			else				offset = DatumGetInt64(d);		}		consumer->offsets[partition] = DatumGetInt64(offset);	}	ExecDropSingleTupleTableSlot(slot);	heap_endscan(scan);	heap_close(offsets, RowExclusiveLock);}

static voidexecute_pg_settings_logger(config_log_objects *objects) {	int		ret;	bool	isnull;	StringInfoData	buf;	SetCurrentStatementStartTimestamp();	StartTransactionCommand();	SPI_connect();	PushActiveSnapshot(GetTransactionSnapshot());	pgstat_report_activity(STATE_RUNNING, "executing configuration logger function");	initStringInfo(&buf);	appendStringInfo(		&buf,		"SELECT %s.%s()",		config_log_schema,		objects->function_name		);	ret = SPI_execute(buf.data, false, 0);	if (ret != SPI_OK_SELECT)	{		elog(FATAL, "SPI_execute failed: error code %d", ret);	}	if (SPI_processed != 1)	{		elog(FATAL, "not a singleton result");	}	log_info("pg_settings_logger() executed");	if(DatumGetBool(SPI_getbinval(SPI_tuptable->vals[0],				  SPI_tuptable->tupdesc,				  1, &isnull)))	{		log_info("Configuration changes recorded");	}	else	{		log_info("No configuration changes detected");	}   	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();	pgstat_report_activity(STATE_IDLE, NULL);}

void worker_main(Datum arg){	int 			ret;	StringInfoData 	buf;	uint32 			segment = UInt32GetDatum(arg);	/* Setup signal handlers */	pqsignal(SIGHUP, worker_sighup);	pqsignal(SIGTERM, worker_sigterm);	/* Allow signals */	BackgroundWorkerUnblockSignals();	initialize_worker(segment);	/* Connect to the database */	BackgroundWorkerInitializeConnection(job->datname, job->rolname);	elog(LOG, "%s initialized running job id %d", MyBgworkerEntry->bgw_name, job->job_id);	pgstat_report_appname(MyBgworkerEntry->bgw_name);	/* Initialize the query text */	initStringInfo(&buf);	appendStringInfo(&buf,					"SELECT * FROM %s.%s(%d, NULL)",					job_run_function.schema,					job_run_function.name,					job->job_id);	/* Initialize the SPI subsystem */	SetCurrentStatementStartTimestamp();	StartTransactionCommand();	SPI_connect();	PushActiveSnapshot(GetTransactionSnapshot());	pgstat_report_activity(STATE_RUNNING, buf.data);	SetCurrentStatementStartTimestamp();	/* And run the query */	ret = SPI_execute(buf.data, true, 0);	if (ret < 0)		elog(FATAL, "errors while executing %s", buf.data);	/* Commmit the transaction */	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();	pgstat_report_activity(STATE_IDLE, NULL);	proc_exit(0);}

/* * This function will set a string in shared memory which is the name of the database to connect to *  the next time the background worker restarts.  Because a bgworker can only connect to one database *  at a time, and some catalogs and stats are scoped to the current database, the bg worker *  periodically restarts to collect latest stats from another database. **/int set_next_db_target(void) {	int retval, processed;	StringInfoData buf;	SPITupleTable *coltuptable;	char* next_db_target;    SetCurrentStatementStartTimestamp();  StartTransactionCommand();  SPI_connect();  PushActiveSnapshot(GetTransactionSnapshot());    /* get sorted list of databases, find one after target_db*/  initStringInfo(&buf);  appendStringInfo(&buf,     "SELECT datname FROM pg_database WHERE datname NOT IN ('template0', 'template1') AND datallowconn IS TRUE AND datname > '%s' ORDER BY datname ASC LIMIT 1;", target_db  );    retval = SPI_execute(buf.data, false, 0);  if (retval != SPI_OK_SELECT) {    elog(FATAL, "Database information collection failed");    // FAIL RETURN 1  }    processed = SPI_processed;    if(processed == 0) {    //No matching records so pick first database.    resetStringInfo(&buf);    appendStringInfoString(&buf,       "SELECT datname FROM pg_database WHERE datname NOT IN ('template0', 'template1') AND datallowconn IS TRUE ORDER BY datname ASC LIMIT 1;"    );    retval = SPI_execute(buf.data, false, 0);    if (retval != SPI_OK_SELECT) {      elog(FATAL, "Database information collection failed");      // FAIL RETURN 1    }  }    coltuptable = SPI_tuptable;    next_db_target = SPI_getvalue(coltuptable->vals[0], coltuptable->tupdesc, 1);  // elog(LOG, "NEXTDB TARGET: %s", next_db_target); //print next target db  strcpy(pgsampler_state->next_db, next_db_target);  	SPI_finish();	PopActiveSnapshot();	CommitTransactionCommand();     return 0;}

/* * GetLatestSnapshot *		Get a snapshot that is up-to-date as of the current instant, *		even if we are executing in transaction-snapshot mode. */SnapshotGetLatestSnapshot(void){	/*	 * So far there are no cases requiring support for GetLatestSnapshot()	 * during logical decoding, but it wouldn't be hard to add if required.	 */	Assert(!HistoricSnapshotActive());	/* If first call in transaction, go ahead and set the xact snapshot */	if (!FirstSnapshotSet)		return GetTransactionSnapshot();	SecondarySnapshot = GetSnapshotData(&SecondarySnapshotData);	return SecondarySnapshot;}

/* * CopyIntoStream * * COPY events to a stream from an input source */voidCopyIntoStream(Relation rel, TupleDesc desc, HeapTuple *tuples, int ntuples){	bool snap = ActiveSnapshotSet();	ResultRelInfo rinfo;	StreamInsertState *sis;	MemSet(&rinfo, 0, sizeof(ResultRelInfo));	rinfo.ri_RangeTableIndex = 1; /* dummy */	rinfo.ri_TrigDesc = NULL;	rinfo.ri_RelationDesc = rel;	if (snap)		PopActiveSnapshot();	BeginStreamModify(NULL, &rinfo, list_make1(desc), 0, 0);	sis = (StreamInsertState *) rinfo.ri_FdwState;	Assert(sis);	if (sis->queries)	{		TupleTableSlot *slot = MakeSingleTupleTableSlot(RelationGetDescr(rel));		int i;		for (i = 0; i < ntuples; i++)		{			ExecStoreTuple(tuples[i], slot, InvalidBuffer, false);			ExecStreamInsert(NULL, &rinfo, slot, NULL);			ExecClearTuple(slot);		}		ExecDropSingleTupleTableSlot(slot);		Assert(sis->ntups == ntuples);		pgstat_increment_cq_write(ntuples, sis->nbytes);	}	EndStreamModify(NULL, &rinfo);	if (snap)		PushActiveSnapshot(GetTransactionSnapshot());}

/* * get_consumer_id * * Get the pipeline_kafka_consumers oid for the given relation-topic pair * */static Oidget_consumer_id(Relation consumers, text *relation, text *topic){	ScanKeyData skey[2];	HeapTuple tup = NULL;	HeapScanDesc scan;	Oid oid = InvalidOid;	ScanKeyInit(&skey[0], 1, BTEqualStrategyNumber, F_TEXTEQ, PointerGetDatum(relation));	ScanKeyInit(&skey[1], 2, BTEqualStrategyNumber, F_TEXTEQ, PointerGetDatum(topic));	scan = heap_beginscan(consumers, GetTransactionSnapshot(), 2, skey);	tup = heap_getnext(scan, ForwardScanDirection);	if (HeapTupleIsValid(tup))		oid = HeapTupleGetOid(tup);	heap_endscan(scan);	return oid;}

/* * enum_endpoint: common code for enum_first/enum_last */static Oidenum_endpoint(Oid enumtypoid, ScanDirection direction){	Relation	enum_rel;	Relation	enum_idx;	SysScanDesc enum_scan;	HeapTuple	enum_tuple;	ScanKeyData skey;	Oid			minmax;	/*	 * Find the first/last enum member using pg_enum_typid_sortorder_index.	 * Note we must not use the syscache, and must use an MVCC snapshot here.	 * See comments for RenumberEnumType in catalog/pg_enum.c for more info.	 */	ScanKeyInit(&skey,				Anum_pg_enum_enumtypid,				BTEqualStrategyNumber, F_OIDEQ,				ObjectIdGetDatum(enumtypoid));	enum_rel = heap_open(EnumRelationId, AccessShareLock);	enum_idx = index_open(EnumTypIdSortOrderIndexId, AccessShareLock);	enum_scan = systable_beginscan_ordered(enum_rel, enum_idx,										   GetTransactionSnapshot(),										   1, &skey);	enum_tuple = systable_getnext_ordered(enum_scan, direction);	if (HeapTupleIsValid(enum_tuple))		minmax = HeapTupleGetOid(enum_tuple);	else		minmax = InvalidOid;	systable_endscan_ordered(enum_scan);	index_close(enum_idx, AccessShareLock);	heap_close(enum_rel, AccessShareLock);	return minmax;}

Datumkafka_consume_begin_all(PG_FUNCTION_ARGS){	HeapTuple tup = NULL;	HeapScanDesc scan;	Relation consumers = open_pipeline_kafka_consumers();	scan = heap_beginscan(consumers, GetTransactionSnapshot(), 0, NULL);	while ((tup = heap_getnext(scan, ForwardScanDirection)) != NULL)	{		Oid id = HeapTupleGetOid(tup);		KafkaConsumer consumer;		load_consumer_state(id, &consumer);		if (!launch_consumer_group(consumers, &consumer, RD_KAFKA_OFFSET_END))			RETURN_FAILURE();	}	heap_endscan(scan);	heap_close(consumers, NoLock);	RETURN_SUCCESS();}

Datumgp_inject_fault(PG_FUNCTION_ARGS){	char	   *faultName = TextDatumGetCString(PG_GETARG_DATUM(0));	char	   *type = TextDatumGetCString(PG_GETARG_DATUM(1));	char	   *ddlStatement = TextDatumGetCString(PG_GETARG_DATUM(2));	char	   *databaseName = TextDatumGetCString(PG_GETARG_DATUM(3));	char	   *tableName = TextDatumGetCString(PG_GETARG_DATUM(4));	int			numOccurrences = PG_GETARG_INT32(5);	int			sleepTimeSeconds = PG_GETARG_INT32(6);	int         dbid = PG_GETARG_INT32(7);	StringInfo  faultmsg = makeStringInfo();	/* Fast path if injecting fault in our postmaster. */	if (GpIdentity.dbid == dbid)	{		appendStringInfo(faultmsg, "%s/n%s/n%s/n%s/n%s/n%d/n%d/n",						 faultName, type, ddlStatement, databaseName,						 tableName, numOccurrences, sleepTimeSeconds);		int offset = 0;		char *response =			processTransitionRequest_faultInject(				faultmsg->data, &offset, faultmsg->len);		if (!response)			elog(ERROR, "failed to inject fault locally (dbid %d)", dbid);		if (strncmp(response, "Success:",  strlen("Success:")) != 0)			elog(ERROR, "%s", response);		elog(NOTICE, "%s", response);		PG_RETURN_DATUM(true);	}	/* Obtain host and port of the requested dbid */	HeapTuple tuple;	Relation rel = heap_open(GpSegmentConfigRelationId, AccessShareLock);	ScanKeyData scankey;	SysScanDesc sscan;	ScanKeyInit(&scankey,				Anum_gp_segment_configuration_dbid,				BTEqualStrategyNumber, F_INT2EQ,				Int16GetDatum((int16) dbid));	sscan = systable_beginscan(rel, GpSegmentConfigDbidIndexId, true,							   GetTransactionSnapshot(), 1, &scankey);	tuple = systable_getnext(sscan);	if (!HeapTupleIsValid(tuple))		elog(ERROR, "cannot find dbid %d", dbid);	bool isnull;	Datum datum = heap_getattr(tuple, Anum_gp_segment_configuration_hostname,							   RelationGetDescr(rel), &isnull);	char *hostname;	if (!isnull)		hostname =				DatumGetCString(DirectFunctionCall1(textout, datum));	else		elog(ERROR, "hostname is null for dbid %d", dbid);	int port = DatumGetInt32(heap_getattr(tuple,										  Anum_gp_segment_configuration_port,										  RelationGetDescr(rel), &isnull));	systable_endscan(sscan);	heap_close(rel, NoLock);	struct addrinfo *addrList = NULL;	struct addrinfo hint;	int			ret;	/* Initialize hint structure */	MemSet(&hint, 0, sizeof(hint));	hint.ai_socktype = SOCK_STREAM;	hint.ai_family = AF_UNSPEC;	char portStr[100];	if (snprintf(portStr, sizeof(portStr), "%d", port) >= sizeof(portStr))		elog(ERROR, "port number too long for dbid %d", dbid);	/* Use pg_getaddrinfo_all() to resolve the address */	ret = pg_getaddrinfo_all(hostname, portStr, &hint, &addrList);	if (ret || !addrList)	{		if (addrList)			pg_freeaddrinfo_all(hint.ai_family, addrList);		elog(ERROR, "could not translate host name /"%s/" to address: %s/n",			 hostname, gai_strerror(ret));	}	PrimaryMirrorTransitionClientInfo client;	client.receivedDataCallbackFn = transitionReceivedDataFn;	client.errorLogFn = transitionErrorLogFn;	client.checkForNeedToExitFn = checkForNeedToExitFn;	transitionMsgErrors = makeStringInfo();	appendStringInfo(faultmsg, "%s/n%s/n%s/n%s/n%s/n%s/n%d/n%d/n",					 "faultInject",	faultName, type, ddlStatement,					 databaseName, tableName, numOccurrences,					 sleepTimeSeconds);	if (sendTransitionMessage(&client, addrList, faultmsg->data, faultmsg->len,							  1 /* retries */, 60 /* timeout */) !=		TRANS_ERRCODE_SUCCESS)//.........这里部分代码省略.........

//.........这里部分代码省略.........     * last thing we do before low-level modules like the buffer manager begin     * to close down.  We need to have this in place before we begin our first     * transaction --- if we fail during the initialization transaction, as is     * entirely possible, we need the AbortTransaction call to clean up.     */    before_shmem_exit(ShutdownPostgres, 0);    /* The autovacuum launcher is done here */    if (IsAutoVacuumLauncherProcess())        return;    /*     * Start a new transaction here before first access to db, and get a     * snapshot.  We don't have a use for the snapshot itself, but we're     * interested in the secondary effect that it sets RecentGlobalXmin. (This     * is critical for anything that reads heap pages, because HOT may decide     * to prune them even if the process doesn't attempt to modify any     * tuples.)     */    if (!bootstrap)    {        /* statement_timestamp must be set for timeouts to work correctly */        SetCurrentStatementStartTimestamp();        StartTransactionCommand();        /*         * transaction_isolation will have been set to the default by the         * above.  If the default is "serializable", and we are in hot         * standby, we will fail if we don't change it to something lower.         * Fortunately, "read committed" is plenty good enough.         */        XactIsoLevel = XACT_READ_COMMITTED;        (void) GetTransactionSnapshot();    }    /*     * Perform client authentication if necessary, then figure out our     * postgres user ID, and see if we are a superuser.     *     * In standalone mode and in autovacuum worker processes, we use a fixed     * ID, otherwise we figure it out from the authenticated user name.     */    if (bootstrap || IsAutoVacuumWorkerProcess())    {        InitializeSessionUserIdStandalone();        am_superuser = true;    }    else if (!IsUnderPostmaster)    {        InitializeSessionUserIdStandalone();        am_superuser = true;        if (!ThereIsAtLeastOneRole())            ereport(WARNING,                    (errcode(ERRCODE_UNDEFINED_OBJECT),                     errmsg("no roles are defined in this database system"),                     errhint("You should immediately run CREATE USER /"%s/" SUPERUSER;.",                             username)));    }    else if (IsBackgroundWorker)    {        if (username == NULL)        {            InitializeSessionUserIdStandalone();            am_superuser = true;        }

/* * CopyIntoStream * * COPY events to a stream from an input source */voidCopyIntoStream(Relation stream, TupleDesc desc, HeapTuple *tuples, int ntuples){	int i;	InsertBatchAck *ack = NULL;	InsertBatch *batch = NULL;	Size size = 0;	bool snap = ActiveSnapshotSet();	Bitmapset *all_targets = GetStreamReaders(RelationGetRelid(stream));	Bitmapset *adhoc = GetAdhocContinuousViewIds();	Bitmapset *targets = bms_difference(all_targets, adhoc);	dsm_cqueue *cq = NULL;	bytea *packed_desc;	if (snap)		PopActiveSnapshot();	packed_desc = PackTupleDesc(desc);	if (!bms_is_empty(targets))	{		if (synchronous_stream_insert)		{			batch = InsertBatchCreate();			ack = palloc0(sizeof(InsertBatchAck));			ack->batch_id = batch->id;			ack->batch = batch;		}		cq = GetWorkerQueue();	}	for (i=0; i<ntuples; i++)	{		StreamTupleState *sts;		HeapTuple tup = tuples[i];		int len;		sts = StreamTupleStateCreate(tup, desc, packed_desc, targets, ack, &len);		if (cq)		{			dsm_cqueue_push_nolock(cq, sts, len);			size += len;		}	}	pfree(packed_desc);	if (cq)		dsm_cqueue_unlock(cq);	stream_stat_report(RelationGetRelid(stream), ntuples, 1, size);	if (batch)	{		pfree(ack);		InsertBatchWaitAndRemove(batch, ntuples);	}	if (snap)		PushActiveSnapshot(GetTransactionSnapshot());	bms_free(all_targets);	bms_free(adhoc);	bms_free(targets);}

//.........这里部分代码省略.........					 "WITH deleted AS (DELETE "					 "FROM %s.%s "					 "WHERE type = 'delta' RETURNING value), "					 "total AS (SELECT coalesce(sum(value), 0) as sum "					 "FROM deleted) "					 "UPDATE %s.%s "					 "SET value = %s.value + total.sum "					 "FROM total WHERE type = 'total' "					 "RETURNING %s.value",					 table->schema, table->name,					 table->schema, table->name,					 table->name,					 table->name);	/*	 * Main loop: do this until the SIGTERM handler tells us to terminate	 */	while (!got_sigterm)	{		int			ret;		int			rc;		/*		 * Background workers mustn't call usleep() or any direct equivalent:		 * instead, they may wait on their process latch, which sleeps as		 * necessary, but is awakened if postmaster dies.  That way the		 * background process goes away immediately in an emergency.		 */		rc = WaitLatch(&MyProc->procLatch,					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,					   worker_spi_naptime * 1000L);		ResetLatch(&MyProc->procLatch);		/* emergency bailout if postmaster has died */		if (rc & WL_POSTMASTER_DEATH)			proc_exit(1);		/*		 * In case of a SIGHUP, just reload the configuration.		 */		if (got_sighup)		{			got_sighup = false;			ProcessConfigFile(PGC_SIGHUP);		}		/*		 * Start a transaction on which we can run queries.  Note that each		 * StartTransactionCommand() call should be preceded by a		 * SetCurrentStatementStartTimestamp() call, which sets both the time		 * for the statement we're about the run, and also the transaction		 * start time.  Also, each other query sent to SPI should probably be		 * preceded by SetCurrentStatementStartTimestamp(), so that statement		 * start time is always up to date.		 *		 * The SPI_connect() call lets us run queries through the SPI manager,		 * and the PushActiveSnapshot() call creates an "active" snapshot		 * which is necessary for queries to have MVCC data to work on.		 *		 * The pgstat_report_activity() call makes our activity visible		 * through the pgstat views.		 */		SetCurrentStatementStartTimestamp();		StartTransactionCommand();		SPI_connect();		PushActiveSnapshot(GetTransactionSnapshot());		pgstat_report_activity(STATE_RUNNING, buf.data);		/* We can now execute queries via SPI */		ret = SPI_execute(buf.data, false, 0);		if (ret != SPI_OK_UPDATE_RETURNING)			elog(FATAL, "cannot select from table %s.%s: error code %d",				 table->schema, table->name, ret);		if (SPI_processed > 0)		{			bool		isnull;			int32		val;			val = DatumGetInt32(SPI_getbinval(SPI_tuptable->vals[0],											  SPI_tuptable->tupdesc,											  1, &isnull));			if (!isnull)				elog(LOG, "%s: count in %s.%s is now %d",					 MyBgworkerEntry->bgw_name,					 table->schema, table->name, val);		}		/*		 * And finish our transaction.		 */		SPI_finish();		PopActiveSnapshot();		CommitTransactionCommand();		pgstat_report_activity(STATE_IDLE, NULL);	}	proc_exit(1);}

//.........这里部分代码省略.........	/*	 * Load relcache entries for the shared system catalogs.  This must create	 * at least an entry for pg_database.	 */	RelationCacheInitializePhase2();	/*	 * Set up process-exit callback to do pre-shutdown cleanup.  This has to	 * be after we've initialized all the low-level modules like the buffer	 * manager, because during shutdown this has to run before the low-level	 * modules start to close down.  On the other hand, we want it in place	 * before we begin our first transaction --- if we fail during the	 * initialization transaction, as is entirely possible, we need the	 * AbortTransaction call to clean up.	 */	on_shmem_exit(ShutdownPostgres, 0);	/* The autovacuum launcher is done here */	if (IsAutoVacuumLauncherProcess())		return;	/*	 * Start a new transaction here before first access to db, and get a	 * snapshot.  We don't have a use for the snapshot itself, but we're	 * interested in the secondary effect that it sets RecentGlobalXmin. (This	 * is critical for anything that reads heap pages, because HOT may decide	 * to prune them even if the process doesn't attempt to modify any	 * tuples.)	 */	if (!bootstrap)	{		StartTransactionCommand();		(void) GetTransactionSnapshot();	}	/*	 * Set up the global variables holding database id and default tablespace.	 * But note we won't actually try to touch the database just yet.	 *	 * We take a shortcut in the bootstrap and walsender case, otherwise we	 * have to look up the db's entry in pg_database.	 */	if (bootstrap || am_walsender)	{		MyDatabaseId = TemplateDbOid;		MyDatabaseTableSpace = DEFAULTTABLESPACE_OID;	}	else if (in_dbname != NULL)	{		HeapTuple	tuple;		Form_pg_database dbform;		tuple = GetDatabaseTuple(in_dbname);		if (!HeapTupleIsValid(tuple))			ereport(FATAL,					(errcode(ERRCODE_UNDEFINED_DATABASE),					 errmsg("database /"%s/" does not exist", in_dbname)));		dbform = (Form_pg_database) GETSTRUCT(tuple);		MyDatabaseId = HeapTupleGetOid(tuple);		MyDatabaseTableSpace = dbform->dattablespace;		/* take database name from the caller, just for paranoia */		strlcpy(dbname, in_dbname, sizeof(dbname));	}	else	{

/* * create_consumer * * Create a row in pipeline_kafka_consumers representing a topic-relation consumer */static Oidcreate_or_update_consumer(Relation consumers, text *relation, text *topic,		text *format, text *delimiter, text *quote, text *escape, int batchsize, int parallelism){	HeapTuple tup;	Datum values[CONSUMER_RELATION_NATTS];	bool nulls[CONSUMER_RELATION_NATTS];	Oid oid;	ScanKeyData skey[2];	HeapScanDesc scan;	MemSet(nulls, false, sizeof(nulls));	ScanKeyInit(&skey[0], 1, BTEqualStrategyNumber, F_TEXTEQ, PointerGetDatum(relation));	ScanKeyInit(&skey[1], 2, BTEqualStrategyNumber, F_TEXTEQ, PointerGetDatum(topic));	scan = heap_beginscan(consumers, GetTransactionSnapshot(), 2, skey);	tup = heap_getnext(scan, ForwardScanDirection);	values[CONSUMER_ATTR_BATCH_SIZE - 1] = Int32GetDatum(batchsize);	values[CONSUMER_ATTR_PARALLELISM - 1] = Int32GetDatum(parallelism);	values[CONSUMER_ATTR_FORMAT - 1] = PointerGetDatum(format);	if (delimiter == NULL)		nulls[CONSUMER_ATTR_DELIMITER - 1] = true;	else		values[CONSUMER_ATTR_DELIMITER - 1] = PointerGetDatum(delimiter);	if (quote == NULL)		nulls[CONSUMER_ATTR_QUOTE - 1] = true;	else		values[CONSUMER_ATTR_QUOTE - 1] = PointerGetDatum(quote);	if (escape == NULL)		nulls[CONSUMER_ATTR_ESCAPE - 1] = true;	else		values[CONSUMER_ATTR_ESCAPE - 1] = PointerGetDatum(escape);	if (HeapTupleIsValid(tup))	{		/* consumer already exists, so just update it with the given parameters */		bool replace[CONSUMER_RELATION_NATTS];		MemSet(replace, true, sizeof(nulls));		replace[CONSUMER_ATTR_RELATION - 1] = false;		replace[CONSUMER_ATTR_TOPIC - 1] = false;		tup = heap_modify_tuple(tup, RelationGetDescr(consumers), values, nulls, replace);		simple_heap_update(consumers, &tup->t_self, tup);		oid = HeapTupleGetOid(tup);	}	else	{		/* consumer doesn't exist yet, create it with the given parameters */		values[CONSUMER_ATTR_RELATION - 1] = PointerGetDatum(relation);		values[CONSUMER_ATTR_TOPIC - 1] = PointerGetDatum(topic);		tup = heap_form_tuple(RelationGetDescr(consumers), values, nulls);		oid = simple_heap_insert(consumers, tup);	}	heap_endscan(scan);	CommandCounterIncrement();	return oid;}

/* * save_consumer_state * * Saves the given consumer's state to pipeline_kafka_consumers */static voidsave_consumer_state(KafkaConsumer *consumer, int partition_group){	ScanKeyData skey[1];	HeapTuple tup = NULL;	HeapScanDesc scan;	Relation offsets = open_pipeline_kafka_offsets();	Datum values[OFFSETS_RELATION_NATTS];	bool nulls[OFFSETS_RELATION_NATTS];	bool replace[OFFSETS_RELATION_NATTS];	bool updated[consumer->num_partitions];	TupleTableSlot *slot = MakeSingleTupleTableSlot(RelationGetDescr(offsets));	int partition;	MemSet(updated, false, sizeof(updated));	ScanKeyInit(&skey[0], OFFSETS_ATTR_CONSUMER, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(consumer->id));	scan = heap_beginscan(offsets, GetTransactionSnapshot(), 1, skey);	/* update any existing offset rows */	while ((tup = heap_getnext(scan, ForwardScanDirection)) != NULL)	{		Datum d;		bool isnull;		int partition;		HeapTuple modified;		ExecStoreTuple(tup, slot, InvalidBuffer, false);		d = slot_getattr(slot, OFFSETS_ATTR_PARTITION, &isnull);		partition = DatumGetInt32(d);		/* we only want to update the offsets we're responsible for */		if (partition % consumer->parallelism != partition_group)			continue;		MemSet(nulls, false, sizeof(nulls));		MemSet(replace, false, sizeof(nulls));		values[OFFSETS_ATTR_OFFSET - 1] = Int64GetDatum(consumer->offsets[partition]);		replace[OFFSETS_ATTR_OFFSET - 1] = true;		updated[partition] = true;		modified = heap_modify_tuple(tup, RelationGetDescr(offsets), values, nulls, replace);		simple_heap_update(offsets, &modified->t_self, modified);	}	heap_endscan(scan);	/* now insert any offset rows that didn't already exist */	for (partition = 0; partition < consumer->num_partitions; partition++)	{		if (updated[partition])			continue;		if (partition % consumer->parallelism != partition_group)			continue;		values[OFFSETS_ATTR_CONSUMER - 1] = ObjectIdGetDatum(consumer->id);		values[OFFSETS_ATTR_PARTITION - 1] = Int32GetDatum(partition);		values[OFFSETS_ATTR_OFFSET - 1] = Int64GetDatum(consumer->offsets[partition]);		MemSet(nulls, false, sizeof(nulls));		tup = heap_form_tuple(RelationGetDescr(offsets), values, nulls);		simple_heap_insert(offsets, tup);	}	ExecDropSingleTupleTableSlot(slot);	heap_close(offsets, NoLock);}