PersistentTidIsKnownResult PersistentStore_TidIsKnown(
	PersistentStoreSharedData 	*storeSharedData,

	ItemPointer 				persistentTid,
	
	ItemPointer 				maxTid)
{
	*maxTid = storeSharedData->maxTid;

	Assert(!PersistentStore_IsZeroTid(persistentTid));

	// UNDONE: I think the InRecovery test only applies to physical Master Mirroring on Standby.
	/* Only test this outside of recovery scenarios */
	if (Persistent_BeforePersistenceWork())
		return PersistentTidIsKnownResult_BeforePersistenceWork;

	if (storeSharedData->needToScanIntoSharedMemory)
		return PersistentTidIsKnownResult_ScanNotPerformedYet;

	if (PersistentStore_IsZeroTid(&storeSharedData->maxTid))
		return PersistentTidIsKnownResult_MaxTidIsZero;

	if (ItemPointerCompare(
						persistentTid,
						&storeSharedData->maxTid) <= 0) // Less-than or equal.
		return PersistentTidIsKnownResult_Known;
	else
		return PersistentTidIsKnownResult_NotKnown;
}

void
PersistentTablespace_ActivateStandby(int16 oldmaster, int16 newmaster)
{
	TablespaceDirEntry tablespaceDirEntry;
	HASH_SEQ_STATUS hstat;

	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	hash_seq_init(&hstat, persistentTablespaceSharedHashTable);

	if (Persistent_BeforePersistenceWork())
		elog(ERROR, "persistent table changes forbidden");

	PersistentTablespace_VerifyInitScan();

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	LWLockAcquire(TablespaceHashLock, LW_SHARED);
	while ((tablespaceDirEntry = hash_seq_search(&hstat)) != NULL)
	{
		PersistentFileSysObjName fsObjName;
		Oid			tblspc = tablespaceDirEntry->key.tablespaceOid;
		ItemPointerData persistentTid;
		uint64		persistentSerialNum;

		tablespaceDirEntry = PersistentTablespace_FindEntryUnderLock(tblspc);

		if (tablespaceDirEntry == NULL)
			elog(ERROR, "cannot find persistent tablespace entry %u",
				 tblspc);

		persistentSerialNum = tablespaceDirEntry->persistentSerialNum;
		ItemPointerCopy(&tablespaceDirEntry->persistentTid, &persistentTid);

		/*
		 * We release TablespaceHashLock in the middle of the loop and
		 * re-acquire it after doing persistent table change.  This is needed
		 * to prevent holding the lock for any purpose other than to protect
		 * the tablespace shared hash table.  Not releasing this lock could
		 * result in file I/O and potential deadlock due to other LW locks
		 * being acquired in the process.  Releasing the lock this way is safe
		 * because we are still holding PersistentObjLock in exclusive mode.
		 * Any change to the filespace shared hash table is also protected by
		 * PersistentObjLock.
		 */

		LWLockRelease(TablespaceHashLock);

		PersistentFileSysObjName_SetTablespaceDir(&fsObjName, tblspc);
		PersistentFileSysObj_ActivateStandby(&fsObjName,
											 &persistentTid,
											 persistentSerialNum,
											 oldmaster,
											 newmaster,
											  /* flushToXlog */ false);
		LWLockAcquire(TablespaceHashLock, LW_SHARED);
	}
	LWLockRelease(TablespaceHashLock);
	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;
}

/*
 * Add a mirror.
 */
void
PersistentFilespace_AddMirror(Oid filespace,
							  char *mirpath,
							  int16 pridbid, int16 mirdbid,
							  bool set_mirror_existence)
{
	PersistentFileSysObjName fsObjName;
	char *newpath;
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	FilespaceDirEntry fde;

	if (Persistent_BeforePersistenceWork())
		elog(ERROR, "persistent table changes forbidden");

	PersistentFilespace_VerifyInitScan();

	PersistentFileSysObjName_SetFilespaceDir(&fsObjName, filespace);

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	fde = PersistentFilespace_FindDirUnderLock(filespace);
	if (fde == NULL)
		elog(ERROR, "did not find persistent filespace entry %u",
			 filespace);

	if (fde->dbId1 == pridbid)
	{
		fde->dbId2 = mirdbid;
		PersistentFilespace_BlankPadCopyLocation(
										fde->locationBlankPadded2,
										mirpath);
		newpath = fde->locationBlankPadded2;
	}
	else if (fde->dbId2 == pridbid)
	{
		fde->dbId1 = mirdbid;
		PersistentFilespace_BlankPadCopyLocation(
										fde->locationBlankPadded1,
										mirpath);
		newpath = fde->locationBlankPadded1;
	}
	else
	{
		Insist(false);
	}

	PersistentFileSysObj_AddMirror(&fsObjName,
								   &fde->persistentTid,
								   fde->persistentSerialNum,
								   pridbid,
								   mirdbid,
								   (void *)newpath,
								   set_mirror_existence,
								   /* flushToXlog */ false);

	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;
}

/*
 * Set mirror for all relation files.
 */
void
PersistentRelation_AddMirrorAll(int16 pridbid, int16 mirdbid)
{
    Relation rel;
    HeapScanDesc scandesc;
    HeapTuple tuple;
    WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

    if (Persistent_BeforePersistenceWork())
        elog(ERROR, "persistent table changes forbidden");

    rel = heap_open(GpPersistentRelationNodeRelationId,
                    AccessExclusiveLock);
    scandesc = heap_beginscan(rel, SnapshotNow, 0, NULL);

    WRITE_PERSISTENT_STATE_ORDERED_LOCK;

    while ((tuple = heap_getnext(scandesc, ForwardScanDirection)) != NULL)
    {
        Form_gp_persistent_relation_node form =
            (Form_gp_persistent_relation_node)GETSTRUCT(tuple);
        Oid tblspcoid = form->tablespace_oid;
        Oid dboid = form->database_oid;
        Oid relfilenode_oid = form->relfilenode_oid;
        int32 segment_file_num = form->segment_file_num;
        int64 serial = form->persistent_serial_num;

        PersistentFileSysObjName fsObjName;
        RelFileNode node;

        node.spcNode = tblspcoid;
        node.dbNode = dboid;
        node.relNode = relfilenode_oid;

        PersistentFileSysObjName_SetRelationFile(&fsObjName,
                &node,
                segment_file_num);

        PersistentFileSysObj_AddMirror(&fsObjName,
                                       &tuple->t_self,
                                       serial,
                                       pridbid,
                                       mirdbid,
                                       NULL,
                                       true,
                                       /* flushToXlog */ false);
    }

    WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;
    heap_endscan(scandesc);
    heap_close(rel, NoLock);
}

bool 
PersistentDatabase_DirIsCreated(DbDirNode *dbDirNode)
{
	READ_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	DatabaseDirEntry databaseDirEntry;
	bool result;

	if (Persistent_BeforePersistenceWork())
	{	
		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(), 
				 "Skipping persistent database '%s' because we are before "
				 "persistence work",
				 GetDatabasePath(
					  dbDirNode->database, 
					  dbDirNode->tablespace));
		/* 
		 * The initdb process will load the persistent table once we out of
		 * bootstrap mode.
		 */
		return true;
	}

	PersistentDatabase_VerifyInitScan();

	READ_PERSISTENT_STATE_ORDERED_LOCK;

	databaseDirEntry =
			(DatabaseDirEntry)
				    SharedOidSearch_Find(
				    		&persistentDatabaseSharedData->databaseDirSearchTable,
				    		dbDirNode->database,
				    		dbDirNode->tablespace);
	result = (databaseDirEntry != NULL);
	if (result &&
		databaseDirEntry->state != PersistentFileSysState_Created &&
		databaseDirEntry->state != PersistentFileSysState_CreatePending &&
		databaseDirEntry->state != PersistentFileSysState_JustInTimeCreatePending)
		elog(ERROR, "Persistent database entry %s expected to be in 'Create Pending' or 'Created' (actual state '%s')", 
			 GetDatabasePath(
				  dbDirNode->database, 
				  dbDirNode->tablespace),
			 PersistentFileSysObjState_Name(databaseDirEntry->state));

	READ_PERSISTENT_STATE_ORDERED_UNLOCK;

	return result;
}

void
PersistentTablespace_RemoveSegment(int16 dbid, bool ismirror)
{
	TablespaceDirEntry tablespaceDirEntry;
	HASH_SEQ_STATUS hstat;

	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	hash_seq_init(&hstat, persistentTablespaceSharedHashTable);

	if (Persistent_BeforePersistenceWork())
		elog(ERROR, "persistent table changes forbidden");

	PersistentTablespace_VerifyInitScan();

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	LWLockAcquire(TablespaceHashLock, LW_SHARED);
	while ((tablespaceDirEntry = hash_seq_search(&hstat)) != NULL)
	{
		PersistentFileSysObjName fsObjName;
		Oid			tblspc = tablespaceDirEntry->key.tablespaceOid;
		ItemPointerData persistentTid;
		uint64		persistentSerialNum;

		tablespaceDirEntry = PersistentTablespace_FindEntryUnderLock(tblspc);

		LWLockRelease(TablespaceHashLock);

		if (tablespaceDirEntry == NULL)
			elog(ERROR, "Did not find persistent tablespace entry %u",
				 tblspc);
		persistentSerialNum = tablespaceDirEntry->persistentSerialNum;
		ItemPointerCopy(&tablespaceDirEntry->persistentTid, &persistentTid);

		PersistentFileSysObjName_SetTablespaceDir(&fsObjName, tblspc);

		PersistentFileSysObj_RemoveSegment(&fsObjName,
										   &persistentTid,
										   persistentSerialNum,
										   dbid,
										   ismirror,
										    /* flushToXlog */ false);
		LWLockAcquire(TablespaceHashLock, LW_SHARED);
	}
	LWLockRelease(TablespaceHashLock);

	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;
}

/*
 * Remove all reference to a segment from the gp_persistent_filespace_node table
 * and share memory structure.
 */
void
PersistentFilespace_RemoveSegment(int16 dbid, bool ismirror)
{
	HASH_SEQ_STATUS hstat;
	FilespaceDirEntry fde;
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	if (Persistent_BeforePersistenceWork())
		elog(ERROR, "persistent table changes forbidden");

	hash_seq_init(&hstat, persistentFilespaceSharedHashTable);

	PersistentFilespace_VerifyInitScan();

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	while ((fde = hash_seq_search(&hstat)) != NULL)
	{
		Oid filespace = fde->key.filespaceOid;
		PersistentFileSysObjName fsObjName;

		PersistentFileSysObjName_SetFilespaceDir(&fsObjName, filespace);

		if (fde->dbId1 == dbid)
		{
			PersistentFilespace_BlankPadCopyLocation(
										fde->locationBlankPadded1,
										"");
		}
		else if (fde->dbId2 == dbid)
		{
			PersistentFilespace_BlankPadCopyLocation(
										fde->locationBlankPadded2,
										"");
		}

		PersistentFileSysObj_RemoveSegment(&fsObjName,
								   &fde->persistentTid,
								   fde->persistentSerialNum,
								   dbid,
								   ismirror,
								   /* flushToXlog */ false);
	}
	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;
}

void PersistentDatabase_MarkJustInTimeCreatePending(
	DbDirNode		*dbDirNode,
	ItemPointer		persistentTid,
	int64			*persistentSerialNum)				
{
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	PersistentFileSysObjName fsObjName;

	DatabaseDirEntry databaseDirEntry;

	SharedOidSearchAddResult addResult;

	if (Persistent_BeforePersistenceWork())
	{	
		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(), 
				 "Skipping persistent database '%s' because we are before persistence work",
				 GetDatabasePath(
					  dbDirNode->database, 
					  dbDirNode->tablespace));

		return;	// The initdb process will load the persistent table once we out of bootstrap mode.
	}

	PersistentDatabase_VerifyInitScan();

	PersistentFileSysObjName_SetDatabaseDir(&fsObjName,dbDirNode->tablespace,dbDirNode->database,is_tablespace_shared);

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	databaseDirEntry =
			(DatabaseDirEntry)
				    SharedOidSearch_Find(
				    		&persistentDatabaseSharedData->databaseDirSearchTable,
				    		dbDirNode->database,
				    		dbDirNode->tablespace);
	if (databaseDirEntry != NULL)
	{
		/*
		 * An existence check should have been done before calling this routine.
		 */
		elog(ERROR, "Persistent database entry '%s' already exists in state '%s'", 
			 GetDatabasePath(
				   dbDirNode->database, 
				   dbDirNode->tablespace),
		     PersistentFileSysObjState_Name(databaseDirEntry->state));
	}

	addResult =
			SharedOidSearch_Add(
					&persistentDatabaseSharedData->databaseDirSearchTable,
					dbDirNode->database,
					dbDirNode->tablespace,
					(SharedOidSearchObjHeader**)&databaseDirEntry);
	if (addResult == SharedOidSearchAddResult_NoMemory)
	{
		/* If out of shared memory, no need to promote to PANIC. */
		WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 errmsg("Out of shared-memory for persistent databases"),
				 errhint("You may need to increase the gp_max_databases and "
				 		 "gp_max_tablespaces value"),
				 errOmitLocation(true)));
	}
	else if (addResult == SharedOidSearchAddResult_Exists)
		elog(PANIC, "Persistent database entry '%s' already exists in state '%s'", 
			 GetDatabasePath(
					dbDirNode->database, 
					dbDirNode->tablespace),
			 PersistentFileSysObjState_Name(databaseDirEntry->state));
	else
		Assert(addResult == SharedOidSearchAddResult_Ok);

	databaseDirEntry->state = PersistentFileSysState_JustInTimeCreatePending;

	PersistentDatabase_AddTuple(
							databaseDirEntry,
							/* reserved */ 0,
							/* parentXid */ InvalidTransactionId,
							/* flushToXLog */ true);


	*persistentTid = databaseDirEntry->persistentTid;
	*persistentSerialNum = databaseDirEntry->persistentSerialNum;

	/*
	 * This XLOG must be generated under the persistent write-lock.
	 */
#ifdef MASTER_MIRROR_SYNC
	mmxlog_log_create_database(dbDirNode->tablespace,
							   dbDirNode->database);	
#endif
	
	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;

}

/*
 * Indicate the non-transaction just-in-time database create was NOT successful.
 */
void PersistentDatabase_AbandonJustInTimeCreatePending(
	DbDirNode 		*dbDirNode,

	ItemPointer 	persistentTid,
				/* TID of the gp_persistent_rel_files tuple for the rel file */

	int64			persistentSerialNum)
				/* Serial number for the relation.	Distinquishes the uses of the tuple. */

{
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	PersistentFileSysObjName fsObjName;

	DatabaseDirEntry databaseDirEntry;

	PersistentFileSysObjStateChangeResult stateChangeResult;
	
	if (Persistent_BeforePersistenceWork())
	{	
		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(), 
				 "Skipping persistent database '%s' because we are before persistence work",
				 GetDatabasePath(
					  dbDirNode->database, 
					  dbDirNode->tablespace));

		return;	// The initdb process will load the persistent table once we out of bootstrap mode.
	}

	PersistentDatabase_VerifyInitScan();

	PersistentFileSysObjName_SetDatabaseDir(&fsObjName,dbDirNode->tablespace,dbDirNode->database,is_tablespace_shared);

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	PersistentDatabase_LookupExistingDbDir(
								dbDirNode,
								&databaseDirEntry);

	if (databaseDirEntry->state != PersistentFileSysState_JustInTimeCreatePending)
		elog(ERROR, "Persistent database entry %s expected to be in 'Just-In-Time Create Pending' state (actual state '%s')", 
			 GetDatabasePath(
			 		dbDirNode->database, 
			 		dbDirNode->tablespace),
			 PersistentFileSysObjState_Name(databaseDirEntry->state));

	stateChangeResult =
		PersistentFileSysObj_StateChange(
								&fsObjName,
								persistentTid,
								persistentSerialNum,
								PersistentFileSysState_Free,
								/* retryPossible */ false,
								/* flushToXlog */ false,
								/* oldState */ NULL,
								/* verifiedActionCallback */ NULL);

	databaseDirEntry->state = PersistentFileSysState_Free;

	if (databaseDirEntry->iteratorRefCount == 0)
		SharedOidSearch_Delete(
					&persistentDatabaseSharedData->databaseDirSearchTable,
					&databaseDirEntry->header);

	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;

	if (Debug_persistent_print)
		elog(Persistent_DebugPrintLevel(), 
		     "Persistent database directory: Abandon '%s' in state 'Just-In-Time Create Pending', serial number " INT64_FORMAT " at TID %s (State-Change result '%s')",
			 PersistentFileSysObjName_ObjectName(&fsObjName),
			 persistentSerialNum,
			 ItemPointerToString(persistentTid),
			 PersistentFileSysObjStateChangeResult_Name(stateChangeResult));
}

/*
 * Indicate we physically removed the relation file.
 */
void PersistentDatabase_Dropped(
	PersistentFileSysObjName *fsObjName,
				/* The tablespace and database OIDs for the dropped relation. */

	ItemPointer		persistentTid,
				/* TID of the gp_persistent_rel_files tuple for the rel file */

	int64			persistentSerialNum)
				/* Serial number for the relation.	Distinquishes the uses of the tuple. */

{
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	DbDirNode 		*dbDirNode = &fsObjName->variant.dbDirNode;

	DatabaseDirEntry databaseDirEntry;

	PersistentFileSysState oldState;

	PersistentFileSysObjStateChangeResult stateChangeResult;
	
	if (Persistent_BeforePersistenceWork())
	{	
		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(), 
				 "Skipping persistent database '%s' because we are before persistence work",
				 GetDatabasePath(
					  dbDirNode->database, 
					  dbDirNode->tablespace));

		return;	// The initdb process will load the persistent table once we out of bootstrap mode.
	}

	PersistentDatabase_VerifyInitScan();

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	PersistentDatabase_LookupExistingDbDir(
								dbDirNode,
								&databaseDirEntry);

	if (databaseDirEntry->state != PersistentFileSysState_DropPending &&
		databaseDirEntry->state != PersistentFileSysState_AbortingCreate)
		elog(ERROR, "Persistent database entry %s expected to be in 'Drop Pending' or 'Aborting Create' (actual state '%s')", 
			 GetDatabasePath(
				  dbDirNode->database, 
				  dbDirNode->tablespace),
			 PersistentFileSysObjState_Name(databaseDirEntry->state));

	stateChangeResult =
		PersistentFileSysObj_StateChange(
								fsObjName,
								persistentTid,
								persistentSerialNum,
								PersistentFileSysState_Free,
								/* retryPossible */ false,
								/* flushToXlog */ false,
								&oldState,
								PersistentDatabase_DroppedVerifiedActionCallback);

	databaseDirEntry->state = PersistentFileSysState_Free;

	if (databaseDirEntry->iteratorRefCount == 0)
		SharedOidSearch_Delete(
					&persistentDatabaseSharedData->databaseDirSearchTable,
					&databaseDirEntry->header);

	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;

	if (Debug_persistent_print)
		elog(Persistent_DebugPrintLevel(), 
		     "Persistent database directory: '%s' changed state from '%s' to (Free), serial number " INT64_FORMAT " at TID %s (State-Change result '%s')",
			 PersistentFileSysObjName_ObjectName(fsObjName),
			 PersistentFileSysObjState_Name(oldState),
			 persistentSerialNum,
			 ItemPointerToString(persistentTid),
			 PersistentFileSysObjStateChangeResult_Name(stateChangeResult));
}

/*
 * Indicate we physically removed the filespace file.
 */
void PersistentFilespace_Dropped(
	Oid 		filespaceOid,
				/* The filespace OID for the dropped filespace. */

	ItemPointer		persistentTid,
				/* TID of the gp_persistent_rel_files tuple for the rel file */

	int64			persistentSerialNum)
				/* Serial number for the filespace.	Distinquishes the uses of the tuple. */

{
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	PersistentFileSysObjName fsObjName;

	FilespaceDirEntry filespaceDirEntry;

	PersistentFileSysState oldState;

	PersistentFileSysObjStateChangeResult stateChangeResult;

	if (Persistent_BeforePersistenceWork())
	{
		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(),
				 "Skipping persistent filespace %u because we are before persistence work",
				 filespaceOid);

		return;	// The initdb process will load the persistent table once we out of bootstrap mode.
	}

	PersistentFilespace_VerifyInitScan();

	PersistentFileSysObjName_SetFilespaceDir(&fsObjName,filespaceOid);

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	stateChangeResult =
		PersistentFileSysObj_StateChange(
								&fsObjName,
								persistentTid,
								persistentSerialNum,
								PersistentFileSysState_Free,
								/* retryPossible */ false,
								/* flushToXlog */ false,
								&oldState,
								PersistentFilespace_DroppedVerifiedActionCallback);

	WRITE_FILESPACE_HASH_LOCK;
	filespaceDirEntry = PersistentFilespace_FindDirUnderLock(filespaceOid);
	if (filespaceDirEntry == NULL)
		elog(ERROR, "Did not find persistent filespace entry %u",
			 filespaceOid);

	if (filespaceDirEntry->state != PersistentFileSysState_DropPending &&
		filespaceDirEntry->state != PersistentFileSysState_AbortingCreate)
		elog(ERROR, "Persistent filespace entry %u expected to be in 'Drop Pending' or 'Aborting Create' (actual state '%s')",
			 filespaceOid,
			 PersistentFileSysObjState_Name(filespaceDirEntry->state));

	filespaceDirEntry->state = PersistentFileSysState_Free;
	PersistentFilespace_RemoveDirUnderLock(filespaceDirEntry);
	WRITE_FILESPACE_HASH_UNLOCK;

	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;

	if (Debug_persistent_print)
		elog(Persistent_DebugPrintLevel(),
		     "Persistent filespace directory: '%s' changed state from '%s' to (Free), serial number " INT64_FORMAT " at TID %s (State-Change result '%s')",
			 PersistentFileSysObjName_ObjectName(&fsObjName),
			 PersistentFileSysObjState_Name(oldState),
			 persistentSerialNum,
			 ItemPointerToString(persistentTid),
			 PersistentFileSysObjStateChangeResult_Name(stateChangeResult));
}

/*
 * Indicate we intend to create a relation file as part of the current transaction.
 *
 * This function adds an entry in 'gp_persistent_relation_node' for either a new table (segment file
 * # 0) or a new segment file under AO table (segment file # > 0 for row/column-oriented AO) with a state
 * 'Create Pending'. An XLOG IntentToCreate record is generated that will guard the subsequent file-system
 * create in case the transaction aborts.
 *
 * Paramaters
 * -----------
 * relFileNode = The tablespace, database, and relation OIDs for the create
 * segmentFileNum = As the name implies (   0 for heap
 *                                       >= 0 for RO/CO AO as applicable)
 * relStorageMgr = Persistent Relation storage Manager
 * relBufpoolKind = Buffer pool type beneath corrosponding relation
 * TODO bufferPollBulkLoad = ???
 * TODO mirrorExistenceState = ???
 * TODO relDataSynchronizationState = ???
 * flushToXlog = If true, the XLOG record for this change will be flushed to disk.
 * TODO isLocalBuf = ???
 *
 * Return
 * ------
 * relationName = Name of the relation used for either debugging or to store in PendingDelete LL.
 * persistentTid = Resulting TID of the gp_persistent_rel_files tuple for the relation
 * serialNum = Resulting serial number for the relation.  Distinquishes the uses of the tuple
 */
void PersistentRelation_AddCreatePending(
    RelFileNode 		*relFileNode,
    int32				segmentFileNum,
    PersistentFileSysRelStorageMgr relStorageMgr,
    PersistentFileSysRelBufpoolKind relBufpoolKind,
    bool				bufferPoolBulkLoad,
    MirroredObjectExistenceState mirrorExistenceState,
    MirroredRelDataSynchronizationState relDataSynchronizationState,
    char				*relationName,
    ItemPointer			persistentTid,
    int64				*serialNum,
    bool 				flushToXLog,
    bool				isLocalBuf)
{
    WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

    PersistentFileSysObjName fsObjName;

    XLogRecPtr mirrorBufpoolResyncCkptLoc;
    ItemPointerData previousFreeTid;

    Datum values[Natts_gp_persistent_relation_node];

    if(RelFileNode_IsEmpty(relFileNode))
        elog(ERROR, "Invalid RelFileNode (0,0,0)");

    MemSet(&previousFreeTid, 0, sizeof(ItemPointerData));
    MemSet(&mirrorBufpoolResyncCkptLoc, 0, sizeof(XLogRecPtr));

    if (Persistent_BeforePersistenceWork())
    {
        if (Debug_persistent_print)
            elog(Persistent_DebugPrintLevel(),
                 "Skipping persistent relation '%s' because we are before persistence work",
                 relpath(*relFileNode));

        MemSet(persistentTid, 0, sizeof(ItemPointerData));
        *serialNum = 0;

        return;	// The initdb process will load the persistent table once we out of bootstrap mode.
    }

    /* Verify if the needed shared mem data structures for persistent tables are setup and inited */
    PersistentRelation_VerifyInitScan();

    /* Setup the file system object name */
    PersistentFileSysObjName_SetRelationFile(
        &fsObjName,
        relFileNode,
        segmentFileNum);

    WRITE_PERSISTENT_STATE_ORDERED_LOCK;

    /* Create a values array which will be used to create a 'gp_persistent_relation_node' tuple */
    GpPersistentRelationNode_SetDatumValues(
        values,
        relFileNode->spcNode,
        relFileNode->dbNode,
        relFileNode->relNode,
        segmentFileNum,
        relStorageMgr,
        (bufferPoolBulkLoad ?
         PersistentFileSysState_BulkLoadCreatePending :
         PersistentFileSysState_CreatePending),
        /* createMirrorDataLossTrackingSessionNum */ 0,
        mirrorExistenceState,
        relDataSynchronizationState,
        /* mirrorBufpoolMarkedForScanIncrementalResync */ false,
        /* mirrorBufpoolResyncChangedPageCount */ 0,
        &mirrorBufpoolResyncCkptLoc,
        /* mirrorBufpoolResyncCkptBlockNum */ 0,
        /* mirrorAppendOnlyLossEof */ 0,
        /* mirrorAppendOnlyNewEof */ 0,
//.........这里部分代码省略.........

PersistentFileSysObjStateChangeResult PersistentRelation_MarkAbortingCreate(
		PersistentFileSysObjName *fsObjName,
		ItemPointer persistentTid,
		int64 persistentSerialNum,
		bool retryPossible)
{
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	RelFileNode *relFileNode = &fsObjName->variant.rel.relFileNode;

	RelationDirEntry relationDirEntry;

	PersistentFileSysObjStateChangeResult stateChangeResult;

	if (RelFileNode_IsEmpty(relFileNode))
	{
		elog(ERROR, "Invalid RelFileNode (0,0,0)");
	}

	if (Persistent_BeforePersistenceWork())
	{
		if (Debug_persistent_print)
		{
			elog(Persistent_DebugPrintLevel(),
				"Skipping persistent relation '%s' because we are before persistence work",
				relpath(*relFileNode));

			/*
			 * The initdb process will load the persistent table once we out of bootstrap mode.
			 */
			return PersistentFileSysObjStateChangeResult_None;
		}
	}

	PersistentRelation_VerifyInitScan();

	/*
	 * Do this check after skipping out if in bootstrap mode.
	 */
	if (PersistentStore_IsZeroTid(persistentTid))
	{
		elog(ERROR, "TID for persistent '%s' tuple for mark DROP pending is invalid (0,0)",
			PersistentFileSysObjName_TypeAndObjectName(fsObjName));
	}

	if (persistentSerialNum == 0)
	{
		elog(ERROR, "Persistent '%s' serial number for mark DROP pending is invalid (0)",
			PersistentFileSysObjName_TypeAndObjectName(fsObjName));
	}

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	relationDirEntry = PersistentRelation_FindEntryUnderLock(relFileNode);

	if (relationDirEntry == NULL)
	{
		elog(ERROR, "Did not find persistent relation entry %u/%u/%u",
			relFileNode->spcNode,
			relFileNode->dbNode,
			relFileNode->relNode);
	}

	if (relationDirEntry->state != PersistentFileSysState_CreatePending)
	{
		elog(ERROR, "Persistent relation entry %u/%u/%u expected to be in 'Create Pending' (actual state '%s')",
			relFileNode->spcNode,
			relFileNode->dbNode,
			relFileNode->relNode,
			PersistentFileSysObjState_Name(relationDirEntry->state));
	}

	stateChangeResult = PersistentFileSysObj_StateChange(
											fsObjName,
											persistentTid,
											persistentSerialNum,
											PersistentFileSysState_AbortingCreate,
											retryPossible,
											/* flushToXLog */ false,
											/* oldState */ NULL,
											/* verifiedActionCallback */ NULL);

	relationDirEntry->state = PersistentFileSysState_AbortingCreate;

	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;

	if (Debug_persistent_print)
	{
		elog(Persistent_DebugPrintLevel(),
			"Persistent relation: '%s' changed state from 'Create Pending' to 'Aborting Create', serial number " INT64_FORMAT " at TID %s (State-Change result '%s')",
			PersistentFileSysObjName_ObjectName(fsObjName),
			persistentSerialNum,
			ItemPointerToString(persistentTid),
			PersistentFileSysObjStateChangeResult_Name(stateChangeResult));
	}

	return stateChangeResult;
}

/*
 * Activate a standby master by removing reference to the dead master
 * and changing our dbid to the old master's dbid
 */
void
PersistentFilespace_ActivateStandby(int16 oldmaster, int16 newmaster)
{
	HASH_SEQ_STATUS hstat;
	FilespaceDirEntry fde;
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	if (Persistent_BeforePersistenceWork())
		elog(ERROR, "persistent table changes forbidden");

	hash_seq_init(&hstat, persistentFilespaceSharedHashTable);

	PersistentFilespace_VerifyInitScan();

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	while ((fde = hash_seq_search(&hstat)) != NULL)
	{
		Oid filespace = fde->key.filespaceOid;
		PersistentFileSysObjName fsObjName;

		PersistentFileSysObjName_SetFilespaceDir(&fsObjName, filespace);

		if (fde->dbId1 == oldmaster)
		{
			fde->dbId1 = InvalidDbid;
			fde->dbId2 = newmaster;

			/* Copy standby filespace location into new master location */
			PersistentFilespace_BlankPadCopyLocation(
										fde->locationBlankPadded2,
										fde->locationBlankPadded1);

			PersistentFilespace_BlankPadCopyLocation(
										fde->locationBlankPadded1,
										"");
		}
		else if (fde->dbId2 == oldmaster)
		{
			fde->dbId2 = InvalidDbid;
			fde->dbId1 = newmaster;

			/* Copy standby filespace location into new master location */
			PersistentFilespace_BlankPadCopyLocation(
										fde->locationBlankPadded1,
										fde->locationBlankPadded2);

			PersistentFilespace_BlankPadCopyLocation(
										fde->locationBlankPadded2,
										"");
		}

		PersistentFileSysObj_ActivateStandby(&fsObjName,
								   &fde->persistentTid,
								   fde->persistentSerialNum,
								   oldmaster,
								   newmaster,
								   /* flushToXlog */ false);
	}

	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;
}

/*
 * Indicate we are aborting the create of a relation file.
 *
 * This state will make sure the relation gets dropped after a system crash.
 */
PersistentFileSysObjStateChangeResult PersistentDatabase_MarkAbortingCreate(
	PersistentFileSysObjName *fsObjName,
				/* The tablespace and database OIDs for the aborting create. */

	ItemPointer		persistentTid,
				/* TID of the gp_persistent_rel_files tuple for the rel file */

	int64			persistentSerialNum,
				/* Serial number for the relation.	Distinquishes the uses of the tuple. */

	bool			retryPossible)
{
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	DbDirNode 		*dbDirNode = &fsObjName->variant.dbDirNode;

	DatabaseDirEntry databaseDirEntry;

	PersistentFileSysObjStateChangeResult stateChangeResult;
	
	if (Persistent_BeforePersistenceWork())
	{	
		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(), 
				 "Skipping persistent database '%s' because we are before persistence work",
				 GetDatabasePath(
					  dbDirNode->database, 
					  dbDirNode->tablespace));

		return false;	// The initdb process will load the persistent table once we out of bootstrap mode.
	}

	PersistentDatabase_VerifyInitScan();

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	PersistentDatabase_LookupExistingDbDir(
								dbDirNode,
								&databaseDirEntry);

	if (databaseDirEntry->state != PersistentFileSysState_CreatePending)
		elog(ERROR, "Persistent database entry %s expected to be in 'Create Pending' (actual state '%s')", 
			 GetDatabasePath(
				   dbDirNode->database, 
				   dbDirNode->tablespace),
			 PersistentFileSysObjState_Name(databaseDirEntry->state));


	stateChangeResult =
		PersistentFileSysObj_StateChange(
								fsObjName,
								persistentTid,
								persistentSerialNum,
								PersistentFileSysState_AbortingCreate,
								retryPossible,
								/* flushToXlog */ false,
								/* oldState */ NULL,
								/* verifiedActionCallback */ NULL);

	databaseDirEntry->state = PersistentFileSysState_AbortingCreate;
		
	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;

	if (Debug_persistent_print)
		elog(Persistent_DebugPrintLevel(), 
		     "Persistent database directory: '%s' changed state from 'Create Pending' to 'Aborting Create', serial number " INT64_FORMAT " at TID %s (State-Change result '%s')",
			 PersistentFileSysObjName_ObjectName(fsObjName),
			 persistentSerialNum,
			 ItemPointerToString(persistentTid),
			 PersistentFileSysObjStateChangeResult_Name(stateChangeResult));

	return stateChangeResult;
}

/*
 * Indicate we are aborting the create of a relation file.
 *
 * This state will make sure the relation gets dropped after a system crash.
 */
PersistentFileSysObjStateChangeResult PersistentRelation_MarkAbortingCreate(
    RelFileNode 		*relFileNode,
    /* The tablespace, database, and relation OIDs for the aborting create. */
    int32				segmentFileNum,
    ItemPointer			persistentTid,
    /* TID of the gp_persistent_rel_files tuple for the relation. */
    int64				persistentSerialNum,
    /* Serial number for the relation.  Distinquishes the uses of the tuple. */
    bool				retryPossible)
{
    WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

    PersistentFileSysObjName fsObjName;

    PersistentFileSysObjStateChangeResult stateChangeResult;

    if(RelFileNode_IsEmpty(relFileNode))
        elog(ERROR, "Invalid RelFileNode (0,0,0)");

    if (Persistent_BeforePersistenceWork())
    {
        if (Debug_persistent_print)
            elog(Persistent_DebugPrintLevel(),
                 "Skipping persistent relation '%s' because we are before persistence work",
                 relpath(*relFileNode));

        return false;	// The initdb process will load the persistent table once we out of bootstrap mode.
    }

    /* MPP-16543: When inserting tuples into AO table, row numbers will be
     * generated from gp_fastsequence catalog table, as part of the design,
     * these sequence numbers are not reusable, even if the AO insert
     * transaction is aborted. The entry in gp_fastsequence was inserted
     * using frozen_heap_insert, which means it's always visible.

     * Aborted AO insert transaction will cause inconsistency between
     * gp_fastsequence and pg_class, the solution is to introduce "frozen
     * delete" - inplace update tuple's MVCC header to make it invisible.
     */

    Relation gp_fastsequence_rel = heap_open(FastSequenceRelationId, RowExclusiveLock);
    HeapTuple   tup;
    SysScanDesc scan;
    ScanKeyData skey;
    ScanKeyInit(&skey,
                Anum_gp_fastsequence_objid,
                BTEqualStrategyNumber,
                F_OIDEQ,
                relFileNode->relNode);

    scan = systable_beginscan(gp_fastsequence_rel,
                              InvalidOid,
                              false,
                              SnapshotNow,
                              1,
                              &skey);
    while (HeapTupleIsValid(tup = systable_getnext(scan)))
    {
        Form_gp_fastsequence found = (Form_gp_fastsequence) GETSTRUCT(tup);
        if (found->objid == relFileNode->relNode)
        {
            if (Debug_persistent_print)
            {
                elog(LOG, "frozen deleting gp_fastsequence entry for aborted AO insert transaction on relation %s", relpath(*relFileNode));
            }

            frozen_heap_inplace_delete(gp_fastsequence_rel, tup);
        }
    }
    systable_endscan(scan);
    heap_close(gp_fastsequence_rel, RowExclusiveLock);




    PersistentRelation_VerifyInitScan();

    PersistentFileSysObjName_SetRelationFile(
        &fsObjName,
        relFileNode,
        segmentFileNum);

    // Do this check after skipping out if in bootstrap mode.
    if (PersistentStore_IsZeroTid(persistentTid))
        elog(ERROR, "TID for persistent '%s' tuple for mark DROP pending is invalid (0,0)",
             PersistentFileSysObjName_TypeAndObjectName(&fsObjName));

    if (persistentSerialNum == 0)
        elog(ERROR, "Persistent '%s' serial number for mark DROP pending is invalid (0)",
             PersistentFileSysObjName_TypeAndObjectName(&fsObjName));

    WRITE_PERSISTENT_STATE_ORDERED_LOCK;

    stateChangeResult =
        PersistentFileSysObj_StateChange(
//.........这里部分代码省略.........

void
PersistentTablespace_AddCreated(
								Oid filespaceOid,
 /* The filespace where the tablespace lives. */

								Oid tablespaceOid,
 /* The tablespace OID to be added. */

								MirroredObjectExistenceState mirrorExistenceState,

								bool flushToXLog)
 /* When true, the XLOG record for this change will be flushed to disk. */

{
	WRITE_PERSISTENT_STATE_ORDERED_LOCK_DECLARE;

	PersistentFileSysObjName fsObjName;

	ItemPointerData persistentTid;
	int64		persistentSerialNum;
	TablespaceDirEntry tablespaceDirEntry;

	if (Persistent_BeforePersistenceWork())
	{
		if (Debug_persistent_print)
			elog(Persistent_DebugPrintLevel(),
				 "Skipping persistent tablespace %u because we are before persistence work",
				 tablespaceOid);

		return;

		/*
		 * The initdb process will load the persistent table once we out of
		 * bootstrap mode.
		 */
	}

	PersistentTablespace_VerifyInitScan();

	PersistentFileSysObjName_SetTablespaceDir(&fsObjName, tablespaceOid);

	WRITE_PERSISTENT_STATE_ORDERED_LOCK;

	PersistentTablespace_AddTuple(
								  filespaceOid,
								  tablespaceOid,
								  PersistentFileSysState_Created,
								   /* createMirrorDataLossTrackingSessionNum */ 0,
								  mirrorExistenceState,
								   /* reserved */ 0,
								  InvalidTransactionId,
								  flushToXLog,
								  &persistentTid,
								  &persistentSerialNum);

	WRITE_TABLESPACE_HASH_LOCK;
	tablespaceDirEntry =
		PersistentTablespace_CreateEntryUnderLock(filespaceOid, tablespaceOid);
	Assert(tablespaceDirEntry != NULL);
	tablespaceDirEntry->state = PersistentFileSysState_Created;
	ItemPointerCopy(&persistentTid, &tablespaceDirEntry->persistentTid);
	tablespaceDirEntry->persistentSerialNum = persistentSerialNum;
	WRITE_TABLESPACE_HASH_UNLOCK;

	WRITE_PERSISTENT_STATE_ORDERED_UNLOCK;

	if (Debug_persistent_print)
		elog(Persistent_DebugPrintLevel(),
			 "Persistent tablespace directory: Add '%s' in state 'Created', mirror existence state '%s', serial number " INT64_FORMAT " at TID '%s' ",
			 PersistentFileSysObjName_ObjectName(&fsObjName),
			 MirroredObjectExistenceState_Name(mirrorExistenceState),
			 persistentSerialNum,
			 ItemPointerToString(&persistentTid));
}