#include "postgres.h"
#include "access/relation.h"
-#include "access/sysattr.h"
#include "access/table.h"
#include "catalog/namespace.h"
#include "catalog/pg_subscription_rel.h"
#include "nodes/makefuncs.h"
#include "replication/logicalrelation.h"
#include "replication/worker_internal.h"
-#include "utils/builtins.h"
#include "utils/inval.h"
-#include "utils/lsyscache.h"
-#include "utils/memutils.h"
-#include "utils/syscache.h"
+
static MemoryContext LogicalRepRelMapContext = NULL;
static HTAB *LogicalRepRelMap = NULL;
-static HTAB *LogicalRepTypMap = NULL;
/*
* Partition map (LogicalRepPartMap)
LogicalRepRelMap = hash_create("logicalrep relation map cache", 128, &ctl,
HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
- /* Initialize the type hash table. */
- MemSet(&ctl, 0, sizeof(ctl));
- ctl.keysize = sizeof(Oid);
- ctl.entrysize = sizeof(LogicalRepTyp);
- ctl.hcxt = LogicalRepRelMapContext;
-
- /* This will usually be small. */
- LogicalRepTypMap = hash_create("logicalrep type map cache", 2, &ctl,
- HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
-
/* Watch for invalidation events. */
CacheRegisterRelcacheCallback(logicalrep_relmap_invalidate_cb,
(Datum) 0);
rel->localrel = NULL;
}
-/*
- * Free the type map cache entry data.
- */
-static void
-logicalrep_typmap_free_entry(LogicalRepTyp *entry)
-{
- pfree(entry->nspname);
- pfree(entry->typname);
-}
-
-/*
- * Add new entry or update existing entry in the type map cache.
- */
-void
-logicalrep_typmap_update(LogicalRepTyp *remotetyp)
-{
- MemoryContext oldctx;
- LogicalRepTyp *entry;
- bool found;
-
- if (LogicalRepTypMap == NULL)
- logicalrep_relmap_init();
-
- /*
- * HASH_ENTER returns the existing entry if present or creates a new one.
- */
- entry = hash_search(LogicalRepTypMap, (void *) &remotetyp->remoteid,
- HASH_ENTER, &found);
-
- if (found)
- logicalrep_typmap_free_entry(entry);
-
- /* Make cached copy of the data */
- entry->remoteid = remotetyp->remoteid;
- oldctx = MemoryContextSwitchTo(LogicalRepRelMapContext);
- entry->nspname = pstrdup(remotetyp->nspname);
- entry->typname = pstrdup(remotetyp->typname);
- MemoryContextSwitchTo(oldctx);
-}
-
-/*
- * Fetch type name from the cache by remote type OID.
- *
- * Return a substitute value if we cannot find the data type; no message is
- * sent to the log in that case, because this is used by error callback
- * already.
- */
-char *
-logicalrep_typmap_gettypname(Oid remoteid)
-{
- LogicalRepTyp *entry;
- bool found;
-
- /* Internal types are mapped directly. */
- if (remoteid < FirstGenbkiObjectId)
- {
- if (!get_typisdefined(remoteid))
- {
- /*
- * This can be caused by having a publisher with a higher
- * PostgreSQL major version than the subscriber.
- */
- return psprintf("unrecognized %u", remoteid);
- }
-
- return format_type_be(remoteid);
- }
-
- if (LogicalRepTypMap == NULL)
- {
- /*
- * If the typemap is not initialized yet, we cannot possibly attempt
- * to search the hash table; but there's no way we know the type
- * locally yet, since we haven't received a message about this type,
- * so this is the best we can do.
- */
- return psprintf("unrecognized %u", remoteid);
- }
-
- /* search the mapping */
- entry = hash_search(LogicalRepTypMap, (void *) &remoteid,
- HASH_FIND, &found);
- if (!found)
- return psprintf("unrecognized %u", remoteid);
-
- Assert(OidIsValid(entry->remoteid));
- return psprintf("%s.%s", entry->nspname, entry->typname);
-}
-
/*
* Partition cache: look up partition LogicalRepRelMapEntry's
*
typedef struct SlotErrCallbackArg
{
LogicalRepRelMapEntry *rel;
- int local_attnum;
int remote_attnum;
} SlotErrCallbackArg;
}
/*
- * Error callback to give more context info about type conversion failure.
+ * Error callback to give more context info about data conversion failures
+ * while reading data from the remote server.
*/
static void
slot_store_error_callback(void *arg)
{
SlotErrCallbackArg *errarg = (SlotErrCallbackArg *) arg;
LogicalRepRelMapEntry *rel;
- char *remotetypname;
- Oid remotetypoid,
- localtypoid;
/* Nothing to do if remote attribute number is not set */
if (errarg->remote_attnum < 0)
return;
rel = errarg->rel;
- remotetypoid = rel->remoterel.atttyps[errarg->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(rel->localreloid, errarg->local_attnum + 1);
-
- errcontext("processing remote data for replication target relation \"%s.%s\" column \"%s\", "
- "remote type %s, local type %s",
+ errcontext("processing remote data for replication target relation \"%s.%s\" column \"%s\"",
rel->remoterel.nspname, rel->remoterel.relname,
- rel->remoterel.attnames[errarg->remote_attnum],
- remotetypname,
- format_type_be(localtypoid));
+ rel->remoterel.attnames[errarg->remote_attnum]);
}
/*
/* Push callback + info on the error context stack */
errarg.rel = rel;
- errarg.local_attnum = -1;
errarg.remote_attnum = -1;
errcallback.callback = slot_store_error_callback;
errcallback.arg = (void *) &errarg;
Oid typinput;
Oid typioparam;
- errarg.local_attnum = i;
errarg.remote_attnum = remoteattnum;
getTypeInputInfo(att->atttypid, &typinput, &typioparam);
typioparam, att->atttypmod);
slot->tts_isnull[i] = false;
- errarg.local_attnum = -1;
errarg.remote_attnum = -1;
}
else
/* For error reporting, push callback + info on the error context stack */
errarg.rel = rel;
- errarg.local_attnum = -1;
errarg.remote_attnum = -1;
errcallback.callback = slot_store_error_callback;
errcallback.arg = (void *) &errarg;
Oid typinput;
Oid typioparam;
- errarg.local_attnum = i;
errarg.remote_attnum = remoteattnum;
getTypeInputInfo(att->atttypid, &typinput, &typioparam);
typioparam, att->atttypmod);
slot->tts_isnull[i] = false;
- errarg.local_attnum = -1;
errarg.remote_attnum = -1;
}
else
/*
* Handle TYPE message.
*
- * Note we don't do local mapping here, that's done when the type is
- * actually used.
+ * This is now vestigial; we read the info and discard it.
*/
static void
apply_handle_type(StringInfo s)
LogicalRepTyp typ;
logicalrep_read_typ(s, &typ);
- logicalrep_typmap_update(&typ);
}
/*
projects / postgresql.git / commitdiff