diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c index 4b2d8a1..e24725c 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -442,8 +442,8 @@ logicalrep_typmap_update(LogicalRepTyp *remotetyp) /* * Fetch type info from the cache. */ -Oid -logicalrep_typmap_getid(Oid remoteid) +char * +logicalrep_typmap_gettypname(Oid remoteid) { LogicalRepTyp *entry; bool found; @@ -454,9 +454,9 @@ logicalrep_typmap_getid(Oid remoteid) { if (!get_typisdefined(remoteid)) ereport(ERROR, - (errmsg("built-in type %u not found", remoteid), + (errmsg("built-in type oid %u not found", remoteid), errhint("This can be caused by having a publisher with a higher PostgreSQL major version than the subscriber."))); - return remoteid; + return format_type_be(remoteid); } if (LogicalRepTypMap == NULL) @@ -467,27 +467,21 @@ logicalrep_typmap_getid(Oid remoteid) HASH_FIND, &found); if (!found) - elog(ERROR, "no type map entry for remote type %u", + elog(ERROR, "no type map entry for remote type oid %u", remoteid); - - /* Found and mapped, return the oid. */ - if (OidIsValid(entry->typoid)) - return entry->typoid; - - /* Otherwise, try to map to local type. */ - nspoid = LookupExplicitNamespace(entry->nspname, true); - if (OidIsValid(nspoid)) - entry->typoid = GetSysCacheOid2(TYPENAMENSP, + + if (!OidIsValid(entry->typoid)) + { + /* Try to map to local type. */ + nspoid = LookupExplicitNamespace(entry->nspname, true); + if (OidIsValid(nspoid)) + entry->typoid = GetSysCacheOid2(TYPENAMENSP, PointerGetDatum(entry->typname), ObjectIdGetDatum(nspoid)); - else - entry->typoid = InvalidOid; - - if (!OidIsValid(entry->typoid)) - ereport(ERROR, - (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("data type \"%s.%s\" required for logical replication does not exist", - entry->nspname, entry->typname))); - - return entry->typoid; + else + entry->typoid = InvalidOid; + } + + /* Return type name */ + return entry->typname; } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 0e68670..b132a23 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -100,8 +100,9 @@ static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping); typedef struct SlotErrCallbackArg { - LogicalRepRelation *rel; - int attnum; + LogicalRepRelMapEntry *rel; + int remote_attnum; + int local_attnum; } SlotErrCallbackArg; static MemoryContext ApplyMessageContext = NULL; @@ -282,17 +283,20 @@ slot_store_error_callback(void *arg) SlotErrCallbackArg *errarg = (SlotErrCallbackArg *) arg; Oid remotetypoid, localtypoid; + char *remotetypname; - if (errarg->attnum < 0) + if (errarg->remote_attnum < 0) return; - remotetypoid = errarg->rel->atttyps[errarg->attnum]; - localtypoid = logicalrep_typmap_getid(remotetypoid); + remotetypoid = errarg->rel->remoterel.atttyps[errarg->remote_attnum]; + remotetypname = logicalrep_typmap_gettypname(remotetypoid); + 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\"", + errarg->rel->remoterel.nspname, errarg->rel->remoterel.relname, + errarg->rel->remoterel.attnames[errarg->remote_attnum], + remotetypname, format_type_be(localtypoid)); } @@ -313,8 +317,9 @@ 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; + errarg.remote_attnum = -1; errcallback.callback = slot_store_error_callback; errcallback.arg = (void *) &errarg; errcallback.previous = error_context_stack; @@ -332,7 +337,8 @@ slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, Oid typinput; Oid typioparam; - errarg.attnum = remoteattnum; + errarg.local_attnum = i; + errarg.remote_attnum = remoteattnum; getTypeInputInfo(att->atttypid, &typinput, &typioparam); slot->tts_values[i] = OidInputFunctionCall(typinput, @@ -378,8 +384,9 @@ 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; + errarg.remote_attnum = -1; errcallback.callback = slot_store_error_callback; errcallback.arg = (void *) &errarg; errcallback.previous = error_context_stack; @@ -402,7 +409,8 @@ slot_modify_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, Oid typinput; Oid typioparam; - errarg.attnum = remoteattnum; + errarg.local_attnum = i; + errarg.remote_attnum = remoteattnum; 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..563bb4f 100644 --- a/src/include/replication/logicalrelation.h +++ b/src/include/replication/logicalrelation.h @@ -37,6 +37,6 @@ extern void logicalrep_rel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode); extern void logicalrep_typmap_update(LogicalRepTyp *remotetyp); -extern Oid logicalrep_typmap_getid(Oid remoteid); +extern char *logicalrep_typmap_gettypname(Oid remoteid); #endif /* LOGICALRELATION_H */