/*
 * 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 void
start_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.
 */
Snapshot
GetLatestSnapshot(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 void
bg_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);
}

void
BeginScanAppendOnlyRelation(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;
}

void
SetEStateSnapshot(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 void
initialize_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 void
apply_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 void
set_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.
 */
Snapshot
GetLatestSnapshot(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 void
load_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 void
execute_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.
 */
Snapshot
GetLatestSnapshot(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
 */
void
CopyIntoStream(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());
}