diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c index e492d26..01ace8f 100644 --- a/src/backend/replication/logical/relation.c +++ b/src/backend/replication/logical/relation.c @@ -35,8 +35,6 @@ static MemoryContext LogicalRepRelMapContext = NULL; static HTAB *LogicalRepRelMap = NULL; static HTAB *LogicalRepTypMap = NULL; -static void logicalrep_typmap_invalidate_cb(Datum arg, int cacheid, - uint32 hashvalue); /* * Relcache invalidation callback for our relation map cache. @@ -115,8 +113,6 @@ logicalrep_relmap_init(void) /* Watch for invalidation events. */ CacheRegisterRelcacheCallback(logicalrep_relmap_invalidate_cb, (Datum) 0); - CacheRegisterSyscacheCallback(TYPEOID, logicalrep_typmap_invalidate_cb, - (Datum) 0); } /* @@ -375,27 +371,6 @@ logicalrep_rel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode) rel->localrel = NULL; } - -/* - * Type cache invalidation callback for our type map cache. - */ -static void -logicalrep_typmap_invalidate_cb(Datum arg, int cacheid, uint32 hashvalue) -{ - HASH_SEQ_STATUS status; - LogicalRepTyp *entry; - - /* Just to be sure. */ - if (LogicalRepTypMap == NULL) - return; - - /* invalidate all cache entries */ - hash_seq_init(&status, LogicalRepTypMap); - - while ((entry = (LogicalRepTyp *) hash_seq_search(&status)) != NULL) - entry->typoid = InvalidOid; -} - /* * Free the type map cache entry data. */ @@ -404,8 +379,6 @@ logicalrep_typmap_free_entry(LogicalRepTyp *entry) { pfree(entry->nspname); pfree(entry->typname); - - entry->typoid = InvalidOid; } /* @@ -436,58 +409,57 @@ logicalrep_typmap_update(LogicalRepTyp *remotetyp) entry->nspname = pstrdup(remotetyp->nspname); entry->typname = pstrdup(remotetyp->typname); MemoryContextSwitchTo(oldctx); - entry->typoid = InvalidOid; } /* - * Fetch type info from the cache. + * Fetch type name from the cache by remote type OID. Return a + * substitute value if we could not found the data type name. + * Since this function is called by slot_store_error_callback + * which is a error context callback function we don't complain + * here. */ -Oid -logicalrep_typmap_getid(Oid remoteid) +char * +logicalrep_typmap_gettypname(Oid remoteid) { LogicalRepTyp *entry; - bool found; - Oid nspoid; + StringInfoData invalid_type_str; + bool found; + + initStringInfo(&invalid_type_str); /* 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 remoteid; + { + /* + * This can be caused by having a publisher with + * a higher PostgreSQL major version than the + * subscriber. + */ + appendStringInfo(&invalid_type_str, "unrecognized built-in type %u", remoteid); + return invalid_type_str.data; + } + + return format_type_be(remoteid); } if (LogicalRepTypMap == NULL) - logicalrep_relmap_init(); + { + appendStringInfo(&invalid_type_str, "unrecognized type %u", remoteid); + return invalid_type_str.data; + } /* 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); - - /* 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, - 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))); + { + appendStringInfo(&invalid_type_str, "unrecognized type %u", remoteid); + return invalid_type_str.data; + } - return entry->typoid; + Assert(OidIsValid(entry->remoteid)); + return entry->typname; } diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 04985c9..c6a9f88 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 local_attnum; + int remote_attnum; } SlotErrCallbackArg; static MemoryContext ApplyMessageContext = NULL; @@ -281,20 +282,30 @@ slot_fill_defaults(LogicalRepRelMapEntry *rel, EState *estate, static void slot_store_error_callback(void *arg) { - SlotErrCallbackArg *errarg = (SlotErrCallbackArg *) arg; + SlotErrCallbackArg *errarg = (SlotErrCallbackArg *) arg; + LogicalRepRelMapEntry *rel; + char *remotetypname; Oid remotetypoid, localtypoid; - if (errarg->attnum < 0) + /* Return if remote attribute number is not set */ + if (errarg->remote_attnum < 0) return; - remotetypoid = errarg->rel->atttyps[errarg->attnum]; - localtypoid = logicalrep_typmap_getid(remotetypoid); + 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(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[errarg->remote_attnum], + remotetypname, format_type_be(localtypoid)); } @@ -315,8 +326,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; @@ -334,8 +346,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, values[remoteattnum], @@ -380,8 +392,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; @@ -404,7 +417,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/logicalproto.h b/src/include/replication/logicalproto.h index 0eb2105..56f6095 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -55,10 +55,9 @@ typedef struct LogicalRepRelation /* Type mapping info */ typedef struct LogicalRepTyp { - Oid remoteid; /* unique id of the type */ - char *nspname; /* schema name */ - char *typname; /* name of the type */ - Oid typoid; /* local type Oid */ + Oid remoteid; /* unique id of the remote type */ + char *nspname; /* schema name of remote type */ + char *typname; /* name of the remote type */ } LogicalRepTyp; /* Transaction info */ diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h index d4250c2..fc2d808 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 */ diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile index 7294b69..a455634 100644 --- a/src/test/modules/Makefile +++ b/src/test/modules/Makefile @@ -17,6 +17,7 @@ SUBDIRS = \ test_rbtree \ test_rls_hooks \ test_shm_mq \ + test_subscription \ worker_spi all: submake-generated-headers diff --git a/src/test/modules/test_subscription/Makefile b/src/test/modules/test_subscription/Makefile new file mode 100644 index 0000000..ec3d13d --- /dev/null +++ b/src/test/modules/test_subscription/Makefile @@ -0,0 +1,19 @@ +# src/test/modules/test_subscription/Makefile + +MODULE_big = test_subscription +OBJS = test_subscription.o $(WIN32RES) +PGFILEDESC = "test_subscription - module for subscription test" + +EXTENSION = test_subscription +DATA = test_subscription--1.0.sql + +ifdef USE_PGXS +PG_CONFIG = pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) +else +subdir = src/test/modules/test_subscription +top_builddir = ../../../.. +include $(top_builddir)/src/Makefile.global +include $(top_srcdir)/contrib/contrib-global.mk +endif diff --git a/src/test/modules/test_subscription/TAGS b/src/test/modules/test_subscription/TAGS new file mode 120000 index 0000000..1a96393 --- /dev/null +++ b/src/test/modules/test_subscription/TAGS @@ -0,0 +1 @@ +/home/masahiko/source/postgresql/TAGS \ No newline at end of file diff --git a/src/test/modules/test_subscription/test_subscription--1.0.sql b/src/test/modules/test_subscription/test_subscription--1.0.sql new file mode 100644 index 0000000..42bc85f --- /dev/null +++ b/src/test/modules/test_subscription/test_subscription--1.0.sql @@ -0,0 +1,22 @@ +/* src/test/modules/test_subscription/test_subscription--1.0.sql */ + +-- complain if script is sourced in psql, rather than via CREATE EXTENSION +\echo Use "CREATE EXTENSION test_subscription" to load this file. \quit + +CREATE TYPE dummytext; + +CREATE FUNCTION dummytext_in(cstring) +RETURNS dummytext +AS 'MODULE_PATHNAME' +LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE; + +CREATE FUNCTION dummytext_out(dummytext) +RETURNS cstring +AS 'MODULE_PATHNAME' +LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE; + +CREATE TYPE dummytext ( + INTERNALLENGTH = -1, + INPUT = dummytext_in, + OUTPUT = dummytext_out +); diff --git a/src/test/modules/test_subscription/test_subscription.c b/src/test/modules/test_subscription/test_subscription.c new file mode 100644 index 0000000..b2578d4 --- /dev/null +++ b/src/test/modules/test_subscription/test_subscription.c @@ -0,0 +1,44 @@ +/*-------------------------------------------------------------------------- + * + * test_subscription.c + * Code for testing logical replication subscriptions. + * + * Copyright (c) 2018, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/test/modules/test_subscription/test_subscription.c + * + * ------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "fmgr.h" + +#include + +PG_MODULE_MAGIC; + +PG_FUNCTION_INFO_V1(dummytext_in); +PG_FUNCTION_INFO_V1(dummytext_out); + +/* Dummy input of data type function */ +Datum +dummytext_in(PG_FUNCTION_ARGS) +{ + char *inputText = PG_GETARG_CSTRING(0); + + if (inputText) + elog(LOG, "intput text: \"%s\"", inputText); + + PG_RETURN_TEXT_P(cstring_to_text(inputText)); +} + +/* Dummy output of data type function */ +Datum +dummytext_out(PG_FUNCTION_ARGS) +{ + Datum txt = PG_GETARG_DATUM(0); + + PG_RETURN_CSTRING(TextDatumGetCString(txt)); +} diff --git a/src/test/modules/test_subscription/test_subscription.control b/src/test/modules/test_subscription/test_subscription.control new file mode 100644 index 0000000..ffea087 --- /dev/null +++ b/src/test/modules/test_subscription/test_subscription.control @@ -0,0 +1,4 @@ +comment = 'Test subscription with an extension' +default_version = '1.0' +module_pathname = '$libdir/test_subscription' +relocatable = true diff --git a/src/test/subscription/Makefile b/src/test/subscription/Makefile index 0f3d209..f5e76bb 100644 --- a/src/test/subscription/Makefile +++ b/src/test/subscription/Makefile @@ -13,7 +13,7 @@ subdir = src/test/subscription top_builddir = ../../.. include $(top_builddir)/src/Makefile.global -EXTRA_INSTALL = contrib/hstore +EXTRA_INSTALL = contrib/hstore src/test/modules/test_subscription check: $(prove_check) diff --git a/src/test/subscription/t/010_slot_store_error.pl b/src/test/subscription/t/010_slot_store_error.pl new file mode 100644 index 0000000..8c4b49f --- /dev/null +++ b/src/test/subscription/t/010_slot_store_error.pl @@ -0,0 +1,56 @@ +# This tests that the errors when data type conversion are correctly +# handled by logical replication apply workers +use strict; +use warnings; +use PostgresNode; +use TestLib; +use Test::More tests => 1; + +# Initialize publisher node +my $node_publisher = get_new_node('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->start; + +# Create subscriber node +my $node_subscriber = get_new_node('subscriber'); +$node_subscriber->init(allows_streaming => 'logical'); +$node_subscriber->start; + +# Setup same table by different steps so that publisher +# and subscriber get different OID of the dummytext data. +# It's necessary for checking if the subscriber can correcly +# look up both remote and local data type strings. +my $ddl = qq( +CREATE EXTENSION test_subscription; +CREATE TABLE test (a dummytext);); + +$node_publisher->safe_psql('postgres', qq( +CREATE EXTENSION hstore;)); +$node_publisher->safe_psql('postgres', $ddl); +$node_subscriber->safe_psql('postgres', $ddl); + +# Setup logical replication +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; +$node_publisher->safe_psql('postgres', qq( +CREATE PUBLICATION tap_pub FOR TABLE test +)); +my $appname = 'tap_sub'; +$node_subscriber->safe_psql('postgres', qq( +CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (slot_name = tap_sub_slot, copy_data = false) +)); + +# Insert test data.which will lead to call the callback +# function for the date type conversion on subscriber. +$node_publisher->safe_psql('postgres', qq( +INSERT INTO test VALUES ('1'); +)); + +$node_publisher->wait_for_catchup($appname); + +# Check the data on subscriber +my $result = $node_subscriber->safe_psql('postgres', qq( +SELECT a FROM test; +)); + +# Inserted data is replicated correctly +is( $result, '1');