diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c index 4b2d8a1..1c23491 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -440,7 +440,42 @@ logicalrep_typmap_update(LogicalRepTyp *remotetyp) } /* - * Fetch type info from the cache. + * Fetch type name from the cache by remote type OID. + */ +char * +logicalrep_typmap_gettypname(Oid remoteid) +{ + LogicalRepTyp *entry; + bool found; + + /* Internal types are mapped directly. */ + if (remoteid < FirstNormalObjectId) + { + if (!get_typisdefined(remoteid)) + ereport(ERROR, + (errmsg("built-in type %u not found", remoteid), + errhint("This can be caused by having a publisher with a higher PostgreSQL major version than the subscriber."))); + return format_type_be(remoteid); + } + + if (LogicalRepTypMap == NULL) + logicalrep_relmap_init(); + + /* Try finding the mapping. */ + entry = hash_search(LogicalRepTypMap, (void *) &remoteid, + HASH_FIND, &found); + + if (!found) + elog(ERROR, "no type map entry for remote type %u", + remoteid); + + Assert(OidIsValid(entry->remoteid)); + + return entry->typname; +} + +/* + * Fetch local type OID from the cache by remote type OID. */ Oid logicalrep_typmap_getid(Oid remoteid) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index fa5d9bb..a51c882 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -100,8 +100,8 @@ static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping); typedef struct SlotErrCallbackArg { - LogicalRepRelation *rel; - int attnum; + LogicalRepRelMapEntry *rel; + int local_attnum; } SlotErrCallbackArg; static MemoryContext ApplyMessageContext = NULL; @@ -282,19 +282,31 @@ static void slot_store_error_callback(void *arg) { SlotErrCallbackArg *errarg = (SlotErrCallbackArg *) arg; + LogicalRepRelMapEntry *rel; + char *remotetypname; + int remote_attnum; Oid remotetypoid, localtypoid; - if (errarg->attnum < 0) + rel = errarg->rel; + remote_attnum = rel->attrmap[errarg->local_attnum]; + + if (remote_attnum < 0) return; - remotetypoid = errarg->rel->atttyps[errarg->attnum]; - localtypoid = logicalrep_typmap_getid(remotetypoid); + remotetypoid = rel->remoterel.atttyps[remote_attnum]; + + /* Fetch remote type name from the LogicalRepTypMap cache */ + remotetypname = logicalrep_typmap_gettypname(remotetypoid); + + /* Fetch local type OID from the local sys cache */ + localtypoid = get_atttype(errarg->rel->localreloid, errarg->local_attnum + 1); + errcontext("processing remote data for replication target relation \"%s.%s\" column \"%s\", " - "remote type %s, local type %s", - errarg->rel->nspname, errarg->rel->relname, - errarg->rel->attnames[errarg->attnum], - format_type_be(remotetypoid), + "remote type \"%s\", local type \"%s\"", + rel->remoterel.nspname, rel->remoterel.relname, + rel->remoterel.attnames[remote_attnum], + remotetypname, format_type_be(localtypoid)); } @@ -315,8 +327,8 @@ slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, ExecClearTuple(slot); /* Push callback + info on the error context stack */ - errarg.rel = &rel->remoterel; - errarg.attnum = -1; + errarg.rel = rel; + errarg.local_attnum = -1; errcallback.callback = slot_store_error_callback; errcallback.arg = (void *) &errarg; errcallback.previous = error_context_stack; @@ -334,7 +346,7 @@ slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, Oid typinput; Oid typioparam; - errarg.attnum = remoteattnum; + errarg.local_attnum = i; getTypeInputInfo(att->atttypid, &typinput, &typioparam); slot->tts_values[i] = OidInputFunctionCall(typinput, @@ -380,8 +392,8 @@ slot_modify_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, ExecClearTuple(slot); /* Push callback + info on the error context stack */ - errarg.rel = &rel->remoterel; - errarg.attnum = -1; + errarg.rel = rel; + errarg.local_attnum = -1; errcallback.callback = slot_store_error_callback; errcallback.arg = (void *) &errarg; errcallback.previous = error_context_stack; @@ -404,7 +416,7 @@ slot_modify_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, Oid typinput; Oid typioparam; - errarg.attnum = remoteattnum; + errarg.local_attnum = i; getTypeInputInfo(att->atttypid, &typinput, &typioparam); slot->tts_values[i] = OidInputFunctionCall(typinput, diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index 8352705..474563c 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -38,5 +38,6 @@ extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel, extern void logicalrep_typmap_update(LogicalRepTyp *remotetyp); extern Oid logicalrep_typmap_getid(Oid remoteid); +extern char *logicalrep_typmap_gettypname(Oid remoteid); #endif /* LOGICALRELATION_H */