From 05789a219a87dccfd9a84c24332d5c655a2f524c Mon Sep 17 00:00:00 2001 From: Dave Cramer Date: Tue, 7 Jul 2020 14:12:20 -0400 Subject: [PATCH 3/4] Actually set the default to unchanged as per the comment Make more readable and fix up whitespace Fix up error messages remove useless options. Add macros for unchanged, binary and text --- .../libpqwalreceiver/libpqwalreceiver.c | 44 +------ src/backend/replication/logical/proto.c | 25 ++-- src/backend/replication/logical/worker.c | 8 +- src/backend/replication/pgoutput/pgoutput.c | 117 ++---------------- src/include/pg_config_manual.h | 2 +- src/include/replication/logicalproto.h | 4 + 6 files changed, 35 insertions(+), 165 deletions(-) diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 08313fa2a5..a571aada71 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -423,52 +423,12 @@ libpqrcv_startstreaming(WalReceiverConn *conn, appendStringInfo(&cmd, ", publication_names %s", pubnames_literal); PQfreemem(pubnames_literal); pfree(pubnames_str); - if (options->proto.logical.binary) { + if (options->proto.logical.binary) + { appendStringInfo(&cmd, ", binary 'true'"); appendStringInfo(&cmd, ", sizeof_datum '%zu'", sizeof(Datum)); appendStringInfo(&cmd, ", sizeof_int '%zu'", sizeof(int)); appendStringInfo(&cmd, ", sizeof_long '%zu'", sizeof(long)); - appendStringInfo(&cmd, ", bigendian '%d'", -#ifdef WORDS_BIGENDIAN - true -#else - false -#endif - ); - appendStringInfo(&cmd, ", float4_byval '%d'", -#if PG_VERSION_NUM >= 130000 - true -#else -#ifdef USE_FLOAT4_BYVAL - true -#else - false -#endif -#endif - ); - appendStringInfo(&cmd, ", float8_byval '%d'", -#ifdef USE_FLOAT8_BYVAL - true -#else - false -#endif - ); - appendStringInfo(&cmd, ", integer_datetimes '%d'", - -/* integer date times are always enabled in version 10 and up */ - -#if PG_VERSION_NUM >= 100000 - true -#else - -#ifdef USE_INTEGER_DATETIMES - true -#else - false -#endif -#endif - - ); } appendStringInfoChar(&cmd, ')'); } diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index 73148f39f3..521f0139a2 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -163,13 +163,16 @@ logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup) char action; LogicalRepRelId relid; + /* read the relation id */ relid = pq_getmsgint(in, 4); + action = pq_getmsgbyte(in); if (action != 'N') elog(ERROR, "expected new tuple but got %d", action); logicalrep_read_tuple(in, newtup); + return relid; } @@ -469,7 +472,7 @@ logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binar } else if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(values[i])) { - pq_sendbyte(out, 'u'); /* unchanged toast column */ + pq_sendbyte(out, LOGICALREP_UNCHANGED); continue; } @@ -486,7 +489,7 @@ logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binar { bytea *outputbytes; int len; - pq_sendbyte(out, 'b'); /* binary send/recv data follows */ + pq_sendbyte(out, LOGICALREP_BINARY); outputbytes = OidSendFunctionCall(typclass->typsend, values[i]); @@ -498,7 +501,7 @@ logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binar } else { - pq_sendbyte(out, 't'); /* 'text' data follows */ + pq_sendbyte(out, LOGICALREP_TEXT); outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]); pq_sendcountedtext(out, outputstr, strlen(outputstr), false); pfree(outputstr); @@ -525,7 +528,7 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple) /* default is unchanged */ tuple->format = palloc(natts * sizeof(char)); - memset(tuple->format, 't', natts * sizeof(char)); + memset(tuple->format, LOGICALREP_UNCHANGED, natts * sizeof(char)); /* Read the data */ @@ -540,21 +543,21 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple) case 'n': /* null */ { tuple->values[i] = (StringInfoData *)NULL; - tuple->format[i] = 't'; + tuple->format[i] = LOGICALREP_TEXT; break; } - case 'u': /* unchanged column */ + case LOGICALREP_UNCHANGED: { /* we don't receive the value of an unchanged column */ tuple->values[i] = (StringInfoData *)NULL; - tuple->format[i] = 'u'; /* be explicit */ + tuple->format[i] = LOGICALREP_UNCHANGED; /* be explicit */ break; } - case 'b': /* binary formatted value */ + case LOGICALREP_BINARY: { int len; StringInfoData *value = palloc(sizeof(StringInfoData)); - tuple->format[i] = 'b'; + tuple->format[i] = LOGICALREP_BINARY; len = pq_getmsgint(in, 4); /* read length */ @@ -570,11 +573,11 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple) tuple->values[i] = value; break; } - case 't': /* text formatted value */ + case LOGICALREP_TEXT: { int len; StringInfoData *value = palloc(sizeof(StringInfoData)); - tuple->format[i] = 't'; + tuple->format[i] = LOGICALREP_TEXT; len = pq_getmsgint(in, 4); /* read length */ /* and data */ diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 806e0a8bed..40952edf18 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -356,7 +356,7 @@ slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, errarg.local_attnum = i; errarg.remote_attnum = remoteattnum; - if (tupleData->format[remoteattnum] == 'b') + if (tupleData->format[remoteattnum] == LOGICALREP_BINARY) { Oid typreceive; Oid typioparam; @@ -465,7 +465,7 @@ slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot, if (remoteattnum < 0) continue; - if (tupleData->format[remoteattnum] =='u') + if (tupleData->format[remoteattnum] == LOGICALREP_UNCHANGED) continue; if (tupleData->values[remoteattnum] != NULL) @@ -474,7 +474,7 @@ slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot, errarg.local_attnum = i; errarg.remote_attnum = remoteattnum; - if (tupleData->format[remoteattnum] == 'b') + if (tupleData->format[remoteattnum] == LOGICALREP_BINARY) { Oid typreceive; Oid typioparam; @@ -820,7 +820,7 @@ apply_handle_update(StringInfo s) target_rte = list_nth(estate->es_range_table, 0); for (int i = 0; i < remoteslot->tts_tupleDescriptor->natts; i++) { - if (newtup.format[i] != 'u') + if (newtup.format[i] != LOGICALREP_UNCHANGED) target_rte->updatedCols = bms_add_member(target_rte->updatedCols, i + 1 - FirstLowInvalidHeapAttributeNumber); } diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 65c2e5d658..837f438e53 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -140,7 +140,6 @@ parse_output_parameters(List *options, uint32 *protocol_version, bool float8_byval; bool integer_datetimes; - // default to false *binary_basetypes = false; @@ -158,7 +157,7 @@ parse_output_parameters(List *options, uint32 *protocol_version, if (protocol_version_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); + errmsg("conflicting or redundant options, \"%s\" already provided", defel->defname))); protocol_version_given = true; if (!scanint8(strVal(defel->arg), true, &parsed)) @@ -179,7 +178,7 @@ parse_output_parameters(List *options, uint32 *protocol_version, if (publication_names_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); + errmsg("conflicting or redundant options, \"%s\" already provided", defel->defname))); publication_names_given = true; if (!SplitIdentifierString(strVal(defel->arg), ',', @@ -194,7 +193,7 @@ parse_output_parameters(List *options, uint32 *protocol_version, if (binary_option_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); + errmsg("conflicting or redundant options, \"%s\" already provided", defel->defname))); binary_option_given = true; if (!parse_bool(strVal(defel->arg), &parsed)) @@ -212,7 +211,7 @@ parse_output_parameters(List *options, uint32 *protocol_version, if (sizeof_datum_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); + errmsg("conflicting or redundant options, \"%s\" already provided", defel->defname))); sizeof_datum_given = true; if (!scanint8(strVal(defel->arg), true, &datum_size)) @@ -226,7 +225,7 @@ parse_output_parameters(List *options, uint32 *protocol_version, if (sizeof_int_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); + errmsg("conflicting or redundant options, \"%s\" already provided", defel->defname))); sizeof_int_given = true; if (!scanint8(strVal(defel->arg), true, &int_size)) @@ -240,7 +239,7 @@ parse_output_parameters(List *options, uint32 *protocol_version, if (sizeof_long_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); + errmsg("conflicting or redundant options, \"%s\" already provided", defel->defname))); sizeof_long_given = true; if (!scanint8(strVal(defel->arg), true, &long_size)) @@ -248,62 +247,6 @@ parse_output_parameters(List *options, uint32 *protocol_version, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("invalid sizeof_long option"))); } - else if (strcmp(defel->defname, "bigendian") == 0) - { - - if (big_endian_given) - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); - big_endian_given = true; - - if (!parse_bool(strVal(defel->arg), &bigendian)) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("invalid bigendian option"))); - } - else if (strcmp(defel->defname, "float4_byval") == 0) - { - - if (float4_byval_given) - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); - float4_byval_given = true; - - if (!parse_bool(strVal(defel->arg), &float4_byval)) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("invalid float4_byval option"))); - } - else if (strcmp(defel->defname, "float8_byval") == 0) - { - - if (float8_byval_given) - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); - float8_byval_given = true; - - if (!parse_bool(strVal(defel->arg), &float8_byval)) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("invalid float8_byval option"))); - } - else if (strcmp(defel->defname, "integer_datetimes") == 0) - { - - if (integer_datetimes_given) - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options %s already provided", defel->defname))); - integer_datetimes_given = true; - - if (!parse_bool(strVal(defel->arg), &integer_datetimes)) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("invalid integer_date_times option"))); - } else elog(ERROR, "unrecognized pgoutput option: %s", defel->defname); } @@ -328,56 +271,16 @@ parse_output_parameters(List *options, uint32 *protocol_version, ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("incompatible long size"))); - if( + #ifdef WORDS_BIGENDIAN - true -#else - false -#endif - != bigendian) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("incompatible endianness"))); - if( float4_byval != -#if PG_VERSION_NUM >= 130000 - true -#else -#ifdef USE_FLOAT4_BYVAL - true + if (bigendian!=true) #else - false -#endif -#endif - ) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("incompatible float4_byval"))); - if( float8_byval != -#ifdef USE_FLOAT8_BYVAL - true -#else - false -#endif - ) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("incompatible float8_byval"))); + if (bigendian!=false) - if ( integer_datetimes != -#if PG_VERSION_NUM >= 100000 - true -#else -#ifdef USE_INTEGER_DATETIMES - true -#else - false #endif -#endif - ) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("incompatible integer_datetimes"))); - + errmsg("incompatible endianness"))); } } diff --git a/src/include/pg_config_manual.h b/src/include/pg_config_manual.h index ec4fa01f30..8f3ec6bde1 100644 --- a/src/include/pg_config_manual.h +++ b/src/include/pg_config_manual.h @@ -336,7 +336,7 @@ * Enable debugging print statements for WAL-related operations; see * also the wal_debug GUC var. */ -#define WAL_DEBUG +/* #define WAL_DEBUG */ /* * Enable tracing of resource consumption during sort operations; diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index b209af4cf2..06b56b253f 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -27,6 +27,10 @@ #define LOGICALREP_PROTO_MIN_VERSION_NUM 1 #define LOGICALREP_PROTO_VERSION_NUM 1 +#define LOGICALREP_UNCHANGED 'u' +#define LOGICALREP_BINARY 'b' +#define LOGICALREP_TEXT 't' + /* Tuple coming via logical replication. */ typedef struct LogicalRepTupleData { -- 2.20.1 (Apple Git-117)