src/backend/replication/logical/reorderbuffer.c
static void
UpdateLogicalMappings(HTAB *tuplecid_data, Oid relid, Snapshot snapshot)
{
	DIR		   *mapping_dir;
	struct dirent *mapping_de;
	List	   *files = NIL;
	ListCell   *file;
	RewriteMappingFile **files_a;
	size_t		off;
	Oid			dboid = IsSharedRelation(relid) ? InvalidOid : MyDatabaseId;

	mapping_dir = AllocateDir("pg_logical/mappings");
	while ((mapping_de = ReadDir(mapping_dir, "pg_logical/mappings")) != NULL)
	{
		Oid			f_dboid;
		Oid			f_relid;
		TransactionId f_mapped_xid;
		TransactionId f_create_xid;
		XLogRecPtr	f_lsn;
		uint32		f_hi,
					f_lo;
		RewriteMappingFile *f;

		if (strcmp(mapping_de->d_name, ".") == 0 ||
			strcmp(mapping_de->d_name, "..") == 0)
			continue;

		/* Ignore files that aren't ours */
		if (strncmp(mapping_de->d_name, "map-", 4) != 0)
			continue;

		if (sscanf(mapping_de->d_name, LOGICAL_REWRITE_FORMAT,
				   &f_dboid, &f_relid, &f_hi, &f_lo,
				   &f_mapped_xid, &f_create_xid) != 6)
			elog(ERROR, "could not parse filename \"%s\"", mapping_de->d_name);

		f_lsn = ((uint64) f_hi) << 32 | f_lo;

		/* mapping for another database */
		if (f_dboid != dboid)
			continue;

		/* mapping for another relation */
		if (f_relid != relid)
			continue;

		/* did the creating transaction abort? */
		if (!TransactionIdDidCommit(f_create_xid))
			continue;

		/* not for our transaction */
		if (!TransactionIdInArray(f_mapped_xid, snapshot->subxip, snapshot->subxcnt))
			continue;

		/* ok, relevant, queue for apply */
		f = palloc(sizeof(RewriteMappingFile));
		f->lsn = f_lsn;
		strcpy(f->fname, mapping_de->d_name);
		files = lappend(files, f);
	}
	FreeDir(mapping_dir);

	/* build array we can easily sort */
	files_a = palloc(list_length(files) * sizeof(RewriteMappingFile *));
	off = 0;
	foreach(file, files)
	{
		files_a[off++] = lfirst(file);
	}

	/* sort files so we apply them in LSN order */
	qsort(files_a, list_length(files), sizeof(RewriteMappingFile *),
		  file_sort_by_lsn);

	for (off = 0; off < list_length(files); off++)
	{
		RewriteMappingFile *f = files_a[off];

		elog(DEBUG1, "applying mapping: \"%s\" in %u", f->fname,
			 snapshot->subxip[0]);
		ApplyLogicalMappingFile(tuplecid_data, relid, f->fname);
		pfree(f);
	}
}
/*
 * Apply any existing logical remapping files if there are any targeted at our
 * transaction for relid.
 */
Apply any existing logical remapping files if there are any targeted at our
transaction for relid.
Apply any existing logical remapping files if there are any targeted at our transaction for relid .
static void
UpdateLogicalMappings(HTAB *tuplecid_data, Oid relid, Snapshot snapshot)
{
	DIR		   *mapping_dir;
	struct dirent *mapping_de;
	List	   *files = NIL;
	ListCell   *file;
	RewriteMappingFile **files_a;
	size_t		off;
	Oid			dboid = IsSharedRelation(relid) ? InvalidOid : MyDatabaseId;
	mapping_dir = AllocateDir("pg_logical/mappings");
	while ((mapping_de = ReadDir(mapping_dir, "pg_logical/mappings")) != NULL)
	{
		Oid			f_dboid;
		Oid			f_relid;
		TransactionId f_mapped_xid;
		TransactionId f_create_xid;
		XLogRecPtr	f_lsn;
		uint32		f_hi,
					f_lo;
		RewriteMappingFile *f;
		if (strcmp(mapping_de->d_name, ".") == 0 ||
			strcmp(mapping_de->d_name, "..") == 0)
			continue;
		if (strncmp(mapping_de->d_name, "map-", 4) != 0)
			continue;
		if (sscanf(mapping_de->d_name, LOGICAL_REWRITE_FORMAT,
				   &f_dboid, &f_relid, &f_hi, &f_lo,
				   &f_mapped_xid, &f_create_xid) != 6)
			elog(ERROR, "could not parse filename \"%s\"", mapping_de->d_name);
		f_lsn = ((uint64) f_hi) << 32 | f_lo;
		if (f_dboid != dboid)
			continue;
		if (f_relid != relid)
			continue;
		if (!TransactionIdDidCommit(f_create_xid))
			continue;
		if (!TransactionIdInArray(f_mapped_xid, snapshot->subxip, snapshot->subxcnt))
			continue;
		f = palloc(sizeof(RewriteMappingFile));
		f->lsn = f_lsn;
		strcpy(f->fname, mapping_de->d_name);
		files = lappend(files, f);
	}
	FreeDir(mapping_dir);
	files_a = palloc(list_length(files) * sizeof(RewriteMappingFile *));
	off = 0;
	foreach(file, files)
	{
		files_a[off++] = lfirst(file);
	}
	qsort(files_a, list_length(files), sizeof(RewriteMappingFile *),
		  file_sort_by_lsn);
	for (off = 0; off < list_length(files); off++)
	{
		RewriteMappingFile *f = files_a[off];
		elog(DEBUG1, "applying mapping: \"%s\" in %u", f->fname,
			 snapshot->subxip[0]);
		ApplyLogicalMappingFile(tuplecid_data, relid, f->fname);
		pfree(f);
	}
}
static void UpdateLogicalMappings ( HTAB * tuplecid_data , Oid relid , Snapshot snapshot ) { DIR * mapping_dir ; struct dirent * mapping_de ; List * files = NIL ; ListCell * file ; RewriteMappingFile * * files_a ; size_t off ; Oid dboid = IsSharedRelation ( relid ) ? InvalidOid : MyDatabaseId ; mapping_dir = AllocateDir ( " " ) ; while ( ( mapping_de = ReadDir ( mapping_dir , " " ) ) != NULL ) { Oid f_dboid ; Oid f_relid ; TransactionId f_mapped_xid ; TransactionId f_create_xid ; XLogRecPtr f_lsn ; uint32 f_hi , f_lo ; RewriteMappingFile * f ; if ( strcmp ( mapping_de -> d_name , " " ) == 0 || strcmp ( mapping_de -> d_name , " " ) == 0 ) continue ; if ( strncmp ( mapping_de -> d_name , " " , 4 ) != 0 ) continue ; if ( sscanf ( mapping_de -> d_name , LOGICAL_REWRITE_FORMAT , & f_dboid , & f_relid , & f_hi , & f_lo , & f_mapped_xid , & f_create_xid ) != 6 ) elog ( ERROR , " \" \" " , mapping_de -> d_name ) ; f_lsn = ( ( uint64 ) f_hi ) << 32 | f_lo ; if ( f_dboid != dboid ) continue ; if ( f_relid != relid ) continue ; if ( ! TransactionIdDidCommit ( f_create_xid ) ) continue ; if ( ! TransactionIdInArray ( f_mapped_xid , snapshot -> subxip , snapshot -> subxcnt ) ) continue ; f = palloc ( sizeof ( RewriteMappingFile ) ) ; f -> lsn = f_lsn ; strcpy ( f -> fname , mapping_de -> d_name ) ; files = lappend ( files , f ) ; } FreeDir ( mapping_dir ) ; files_a = palloc ( list_length ( files ) * sizeof ( RewriteMappingFile * ) ) ; off = 0 ; foreach ( file , files ) { files_a [ off ++ ] = lfirst ( file ) ; } qsort ( files_a , list_length ( files ) , sizeof ( RewriteMappingFile * ) , file_sort_by_lsn ) ; for ( off = 0 ; off < list_length ( files ) ; off ++ ) { RewriteMappingFile * f = files_a [ off ] ; elog ( DEBUG1 , " \" \" " , f -> fname , snapshot -> subxip [ 0 ] ) ; ApplyLogicalMappingFile ( tuplecid_data , relid , f -> fname ) ; pfree ( f ) ; } }
Apply any existing logical remapping files if there are any targeted at our
 transaction for relid.
Apply any existing logical remapping files if there are any targeted at our transaction for relid .
/* Ignore files that aren't ours */ /* mapping for another database */ /* mapping for another relation */ /* did the creating transaction abort? */ /* not for our transaction */ /* ok, relevant, queue for apply */ /* build array we can easily sort */ /* sort files so we apply them in LSN order */