From 59db2a14cc30cf1017a1f49c51f656dbf3d7923e Mon Sep 17 00:00:00 2001 From: xue_meng_en <1836611252@qq.com> Date: Thu, 31 Mar 2022 14:55:10 +0800 Subject: [PATCH] =?UTF-8?q?Revert=20"=E5=8F=91=E5=B8=83=E8=AE=A2=E9=98=85?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E4=BB=A5=E4=BA=8C=E8=BF=9B=E5=88=B6=E6=A0=BC?= =?UTF-8?q?=E5=BC=8F=E5=8F=91=E9=80=81=E6=95=B0=E6=8D=AE"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit 71f53e575c72d405bae163eedadf13fd64cff85e. --- src/bin/pg_dump/pg_dump.cpp | 31 ++-- src/bin/pg_dump/pg_dump.h | 1 - .../backend/catalog/pg_subscription.cpp | 4 - src/common/backend/utils/init/globals.cpp | 3 +- .../optimizer/commands/subscriptioncmds.cpp | 28 +--- .../storage/replication/libpqwalreceiver.cpp | 8 +- .../storage/replication/logical/proto.cpp | 128 +++++++---------- .../storage/replication/logical/worker.cpp | 136 +++++------------- .../storage/replication/pgoutput/pgoutput.cpp | 25 +--- src/include/catalog/pg_subscription.h | 6 +- src/include/miscadmin.h | 1 - src/include/replication/libpqwalreceiver.h | 1 - src/include/replication/logicalproto.h | 19 +-- src/include/replication/pgoutput.h | 1 - src/test/regress/input/subscription.source | 5 +- src/test/regress/output/subscription.source | 19 +-- 16 files changed, 114 insertions(+), 302 deletions(-) diff --git a/src/bin/pg_dump/pg_dump.cpp b/src/bin/pg_dump/pg_dump.cpp index a6d00d50..7e0a942a 100644 --- a/src/bin/pg_dump/pg_dump.cpp +++ b/src/bin/pg_dump/pg_dump.cpp @@ -233,7 +233,6 @@ char* all_data_nodename_list = NULL; const uint32 USTORE_UPGRADE_VERSION = 92368; const uint32 PACKAGE_ENHANCEMENT = 92444; const uint32 SUBSCRIPTION_VERSION = 92580; -const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92607; #ifdef DUMPSYSLOG char* syslogpath = NULL; @@ -4445,9 +4444,7 @@ void getSubscriptions(Archive *fout) int i_subslotname; int i_subsynccommit; int i_subpublications; - int i_subbinary; - int i; - int ntups; + int i, ntups; if (no_subscriptions || GetVersionNum(fout) < SUBSCRIPTION_VERSION) { return; @@ -4472,20 +4469,14 @@ void getSubscriptions(Archive *fout) resetPQExpBuffer(query); /* Get the subscriptions in current database. */ - appendPQExpBuffer(query, "SELECT s.tableoid, s.oid, s.subname," - "(%s s.subowner) AS rolname, s.subconninfo, s.subslotname, " - "s.subsynccommit, s.subpublications, \n", username_subquery); - - if (GetVersionNum(fout) >= SUBSCRIPTION_BINARY_VERSION_NUM) { - appendPQExpBuffer(query, " s.subbinary\n"); - } else { - appendPQExpBuffer(query, " false AS subbinary\n"); - } - - appendPQExpBuffer(query, "FROM pg_catalog.pg_subscription s " + appendPQExpBuffer(query, + "SELECT s.tableoid, s.oid, s.subname," + "(%s s.subowner) AS rolname, " + " s.subconninfo, s.subslotname, s.subsynccommit, s.subpublications " + "FROM pg_catalog.pg_subscription s " "WHERE s.subdbid = (SELECT oid FROM pg_catalog.pg_database" - " WHERE datname = current_database())"); - + " WHERE datname = current_database())", + username_subquery); res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK); ntups = PQntuples(res); @@ -4503,7 +4494,6 @@ void getSubscriptions(Archive *fout) i_subslotname = PQfnumber(res, "subslotname"); i_subsynccommit = PQfnumber(res, "subsynccommit"); i_subpublications = PQfnumber(res, "subpublications"); - i_subbinary = PQfnumber(res, "subbinary"); subinfo = (SubscriptionInfo *)pg_malloc(ntups * sizeof(SubscriptionInfo)); @@ -4522,7 +4512,6 @@ void getSubscriptions(Archive *fout) } subinfo[i].subsynccommit = gs_strdup(PQgetvalue(res, i, i_subsynccommit)); subinfo[i].subpublications = gs_strdup(PQgetvalue(res, i, i_subpublications)); - subinfo[i].subbinary = gs_strdup(PQgetvalue(res, i, i_subbinary)); if (strlen(subinfo[i].rolname) == 0) { write_msg(NULL, "WARNING: owner of subscription \"%s\" appears to be invalid\n", subinfo[i].dobj.name); @@ -4589,10 +4578,6 @@ static void dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo) appendPQExpBufferStr(query, "NONE"); } - if (strcmp(subinfo->subbinary, "t") == 0) { - appendPQExpBuffer(query, ", binary = true"); - } - if (strcmp(subinfo->subsynccommit, "off") != 0) { appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit)); } diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index a705192c..4c7fdbc9 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -498,7 +498,6 @@ typedef struct _SubscriptionInfo { char *subslotname; char *subsynccommit; char *subpublications; - char *subbinary; } SubscriptionInfo; /* global decls */ diff --git a/src/common/backend/catalog/pg_subscription.cpp b/src/common/backend/catalog/pg_subscription.cpp index f2d31b21..3d1d6833 100644 --- a/src/common/backend/catalog/pg_subscription.cpp +++ b/src/common/backend/catalog/pg_subscription.cpp @@ -91,10 +91,6 @@ Subscription *GetSubscription(Oid subid, bool missing_ok) } sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum)); - datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, Anum_pg_subscription_subbinary, &isnull); - Assert(!isnull); - sub->binary = DatumGetBool(datum); - ReleaseSysCache(tup); return sub; diff --git a/src/common/backend/utils/init/globals.cpp b/src/common/backend/utils/init/globals.cpp index d6014e6d..8b4b9cf8 100644 --- a/src/common/backend/utils/init/globals.cpp +++ b/src/common/backend/utils/init/globals.cpp @@ -59,7 +59,7 @@ bool open_join_children = true; bool will_shutdown = false; /* hard-wired binary version number */ -const uint32 GRAND_VERSION_NUM = 92607; +const uint32 GRAND_VERSION_NUM = 92606; const uint32 PREDPUSH_SAME_LEVEL_VERSION_NUM = 92522; const uint32 UPSERT_WHERE_VERSION_NUM = 92514; @@ -101,7 +101,6 @@ const uint32 PRIVS_DIRECTORY_VERSION_NUM = 92460; const uint32 COMMENT_RECORD_PARAM_VERSION_NUM = 92484; const uint32 SCAN_BATCH_MODE_VERSION_NUM = 92568; const uint32 PUBLICATION_VERSION_NUM = 92580; -const uint32 SUBSCRIPTION_BINARY_VERSION_NUM = 92607; /* Version number of the guc parameter backend_version added in V500R001C20 */ const uint32 V5R1C20_BACKEND_VERSION_NUM = 92305; diff --git a/src/gausskernel/optimizer/commands/subscriptioncmds.cpp b/src/gausskernel/optimizer/commands/subscriptioncmds.cpp index 15ced847..b40214c0 100644 --- a/src/gausskernel/optimizer/commands/subscriptioncmds.cpp +++ b/src/gausskernel/optimizer/commands/subscriptioncmds.cpp @@ -56,7 +56,7 @@ static void ValidateReplicationSlot(char *slotname, List *publications); * accommodate that. */ static void parse_subscription_options(const List *options, char **conninfo, List **publications, bool *enabled_given, - bool *enabled, bool *slot_name_given, char **slot_name, char **synchronous_commit, bool *binary_given, bool *binary) + bool *enabled, bool *slot_name_given, char **slot_name, char **synchronous_commit) { ListCell *lc; @@ -76,10 +76,6 @@ static void parse_subscription_options(const List *options, char **conninfo, Lis if (synchronous_commit) { *synchronous_commit = NULL; } - if (binary) { - *binary_given = false; - *binary = false; - } /* Parse options */ foreach (lc, options) { @@ -128,15 +124,6 @@ static void parse_subscription_options(const List *options, char **conninfo, Lis /* Test if the given value is valid for synchronous_commit GUC. */ (void)set_config_option("synchronous_commit", *synchronous_commit, PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET, false, 0, false); - } else if (strcmp(defel->defname, "binary") == 0 && binary) { - if (*binary_given) { - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("conflicting or redundant options"))); - } - - *binary_given = true; - *binary = defGetBoolean(defel); } else { ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("unrecognized subscription parameter: %s", defel->defname))); @@ -309,8 +296,6 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) char *conninfo; char *slotname; bool slotname_given; - bool binary; - bool binary_given; char originname[NAMEDATALEN]; List *publications; int rc; @@ -320,7 +305,7 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) * Connection and publication should not be specified here. */ parse_subscription_options(stmt->options, NULL, NULL, &enabled_given, &enabled, &slotname_given, &slotname, - &synchronous_commit, &binary_given, &binary); + &synchronous_commit); /* * Since creating a replication slot is not transactional, rolling back @@ -364,7 +349,6 @@ ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) values[Anum_pg_subscription_subname - 1] = DirectFunctionCall1(namein, CStringGetDatum(stmt->subname)); values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(owner); values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled); - values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(binary); /* encrypt conninfo */ List *conninfoList = ConninfoToDefList(stmt->conninfo); @@ -455,8 +439,6 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt) Oid subid; bool enabled_given = false; bool enabled; - bool binary_given; - bool binary; char *synchronous_commit; char *conninfo; char *slot_name; @@ -491,7 +473,7 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt) /* Parse options. */ parse_subscription_options(stmt->options, &conninfo, &publications, &enabled_given, &enabled, &slotname_given, - &slot_name, &synchronous_commit, &binary_given, &binary); + &slot_name, &synchronous_commit); /* Form a new tuple. */ rc = memset_s(nulls, sizeof(nulls), false, sizeof(nulls)); @@ -566,10 +548,6 @@ ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt) values[Anum_pg_subscription_subsynccommit - 1] = CStringGetTextDatum(synchronous_commit); replaces[Anum_pg_subscription_subsynccommit - 1] = true; } - if (binary_given) { - values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(binary); - replaces[Anum_pg_subscription_subbinary - 1] = true; - } if (publications != NIL) { values[Anum_pg_subscription_subpublications - 1] = publicationListToArray(publications); replaces[Anum_pg_subscription_subpublications - 1] = true; diff --git a/src/gausskernel/storage/replication/libpqwalreceiver.cpp b/src/gausskernel/storage/replication/libpqwalreceiver.cpp index 22cf77ee..41bab8a7 100755 --- a/src/gausskernel/storage/replication/libpqwalreceiver.cpp +++ b/src/gausskernel/storage/replication/libpqwalreceiver.cpp @@ -255,14 +255,8 @@ void StartRemoteStreaming(const LibpqrcvConnectParam *options) stringlist_to_identifierstr(t_thrd.libwalreceiver_cxt.streamConn, options->publicationNames); appendStringInfo(&cmd, ", publication_names %s", PQescapeLiteral(t_thrd.libwalreceiver_cxt.streamConn, pubnames_str, strlen(pubnames_str))); - pfree(pubnames_str); - - if (options->binary && PQserverVersion(t_thrd.libwalreceiver_cxt.streamConn) >= 90204) { - appendStringInfoString(&cmd, ", binary 'true'"); - ereport(DEBUG5, ( errmsg("append binary true"))); - } - appendStringInfoChar(&cmd, ')'); + pfree(pubnames_str); } PGresult *res = libpqrcv_PQexec(cmd.data); diff --git a/src/gausskernel/storage/replication/logical/proto.cpp b/src/gausskernel/storage/replication/logical/proto.cpp index c8d993ca..5387fe2a 100644 --- a/src/gausskernel/storage/replication/logical/proto.cpp +++ b/src/gausskernel/storage/replication/logical/proto.cpp @@ -18,6 +18,7 @@ #include "catalog/pg_type.h" #include "libpq/pqformat.h" #include "replication/logicalproto.h" +#include "utils/builtins.h" #include "utils/lsyscache.h" #include "utils/syscache.h" @@ -27,7 +28,7 @@ static const int LOGICALREP_IS_REPLICA_IDENTITY = 1; static void logicalrep_write_attrs(StringInfo out, Relation rel); -static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binary); +static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple); static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel); static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple); @@ -114,7 +115,7 @@ void logicalrep_write_origin(StringInfo out, const char *origin, XLogRecPtr orig /* * Write INSERT to the output stream. */ -void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple, bool binary) +void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple) { pq_sendbyte(out, 'I'); /* action INSERT */ @@ -122,7 +123,7 @@ void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple, b pq_sendint32(out, RelationGetRelid(rel)); pq_sendbyte(out, 'N'); /* new tuple follows */ - logicalrep_write_tuple(out, rel, newtuple, binary); + logicalrep_write_tuple(out, rel, newtuple); } /* @@ -150,7 +151,7 @@ LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtu /* * Write UPDATE to the output stream. */ -void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple, bool binary) +void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple) { pq_sendbyte(out, 'U'); /* action UPDATE */ @@ -165,11 +166,11 @@ void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, H pq_sendbyte(out, 'O'); /* old tuple follows */ else pq_sendbyte(out, 'K'); /* old key follows */ - logicalrep_write_tuple(out, rel, oldtuple, binary); + logicalrep_write_tuple(out, rel, oldtuple); } pq_sendbyte(out, 'N'); /* new tuple follows */ - logicalrep_write_tuple(out, rel, newtuple, binary); + logicalrep_write_tuple(out, rel, newtuple); } /* @@ -212,7 +213,7 @@ LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, Logica /* * Write DELETE to the output stream. */ -void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple, bool binary) +void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple) { char relreplident = RelationGetRelReplident(rel); Assert(relreplident == REPLICA_IDENTITY_DEFAULT || @@ -228,7 +229,7 @@ void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple, b else pq_sendbyte(out, 'K'); /* old key follows */ - logicalrep_write_tuple(out, rel, oldtuple, binary); + logicalrep_write_tuple(out, rel, oldtuple); } /* @@ -343,7 +344,7 @@ void logicalrep_read_typ(StringInfo in, LogicalRepTyp *ltyp) /* * Write a tuple to the outputstream, in the most efficient format possible. */ -static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple, bool binary) +static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple) { TupleDesc desc; Datum values[MaxTupleAttributeNumber]; @@ -370,6 +371,7 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple HeapTuple typtup; Form_pg_type typclass; Form_pg_attribute att = desc->attrs[i]; + char *outputstr; /* skip dropped columns */ if (att->attisdropped || GetGeneratedCol(desc, i)) { @@ -377,7 +379,7 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple } if (isnull[i]) { - pq_sendbyte(out, LOGICALREP_COLUMN_NULL); /* null column */ + pq_sendbyte(out, 'n'); /* null column */ continue; } @@ -386,93 +388,61 @@ static void logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple elog(ERROR, "cache lookup failed for type %u", att->atttypid); typclass = (Form_pg_type)GETSTRUCT(typtup); - /* - * Send in binary if requested and type has suitable send function. - */ - if (binary && OidIsValid(typclass->typsend)) { - bytea* outputbytes = NULL; - pq_sendbyte(out, LOGICALREP_COLUMN_BINARY); - if (!typclass->typbyval && typclass->typlen == -1) { - /* definitely detoasted Datum */ - Datum val = PointerGetDatum(PG_DETOAST_DATUM(values[i])); - outputbytes = OidSendFunctionCall(typclass->typsend, val); - } else { - outputbytes = OidSendFunctionCall(typclass->typsend, values[i]); - } - int len = VARSIZE(outputbytes) - VARHDRSZ; - pq_sendint(out, len, 4); /* length */ - pq_sendbytes(out, VARDATA(outputbytes), len); /* data */ - if (outputbytes != NULL) { - pfree(outputbytes); - } + pq_sendbyte(out, 't'); /* 'text' data follows */ + if (!typclass->typbyval && typclass->typlen == -1) { + /* definitely detoasted Datum */ + Datum val = PointerGetDatum(PG_DETOAST_DATUM(values[i])); + outputstr = OidOutputFunctionCall(typclass->typoutput, val); } else { - char *outputstr; - pq_sendbyte(out, LOGICALREP_COLUMN_TEXT); - if (!typclass->typbyval && typclass->typlen == -1) { - /* definitely detoasted Datum */ - Datum val = PointerGetDatum(PG_DETOAST_DATUM(values[i])); - outputstr = OidOutputFunctionCall(typclass->typoutput, val); - } else { - outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]); - } - pq_sendcountedtext(out, outputstr, strlen(outputstr), false); - if (outputstr != NULL) { - pfree(outputstr); - } + outputstr = OidOutputFunctionCall(typclass->typoutput, values[i]); } + pq_sendcountedtext(out, outputstr, strlen(outputstr), false); + pfree(outputstr); ReleaseSysCache(typtup); } } /* - * Read tuple in logical replication format from stream. + * Read tuple in remote format from stream. + * + * The returned tuple points into the input stringinfo. */ static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple) { - /* Get number of attributes. */ - uint16 natts = pq_getmsgint(in, sizeof(uint16)); + uint16 i; + uint16 natts; + int rc; - /* Allocate space for per-column values; zero out unused StringInfoDatas */ - tuple->colvalues = (StringInfoData *) palloc0(natts * sizeof(StringInfoData)); - tuple->colstatus = (char *) palloc(natts * sizeof(char)); + /* Get number of attributes. */ + natts = pq_getmsgint(in, sizeof(uint16)); + + rc = memset_s(tuple->changed, sizeof(tuple->changed), 0, sizeof(tuple->changed)); + securec_check(rc, "", ""); tuple->ncols = natts; /* Read the data */ - for (uint16 i = 0; i < natts; i++) { - char kind = pq_getmsgbyte(in); - tuple->colstatus[i] = kind; - uint32 len; - StringInfo value = &tuple->colvalues[i]; + for (i = 0; i < natts; i++) { + char kind; + + kind = pq_getmsgbyte(in); switch (kind) { - case LOGICALREP_COLUMN_NULL: /* null */ - /* nothing more to do */ + case 'n': /* null */ + tuple->values[i] = NULL; + tuple->changed[i] = true; break; - case LOGICALREP_COLUMN_TEXT: - len = pq_getmsgint(in, sizeof(uint32)); /* read length */ + case 't': { /* text formatted value */ + uint32 len; + tuple->changed[i] = true; + + len = pq_getmsgint(in, sizeof(uint32)); /* read length */ /* and data */ - value->data = (char *) palloc((len + 1) * sizeof(char)); - pq_copymsgbytes(in, value->data, len); - value->data[len] = '\0'; - /* make StringInfo fully valid */ - value->len = len; - value->cursor = 0; - value->maxlen = len; - break; - case LOGICALREP_COLUMN_BINARY: - len = pq_getmsgint(in, sizeof(uint32)); /* read length */ - - /* and data */ - value->data = (char *) palloc((len + 1) * sizeof(char)); - pq_copymsgbytes(in, value->data, len); - /* not strictly necessary but per StringInfo practice */ - value->data[len] = '\0'; - /* make StringInfo fully valid */ - value->len = len; - value->cursor = 0; - value->maxlen = len; + tuple->values[i] = (char *)palloc(len + 1); + pq_copymsgbytes(in, tuple->values[i], len); + tuple->values[i][len] = '\0'; break; + } default: elog(ERROR, "unrecognized data representation type '%c'", kind); break; @@ -481,7 +451,7 @@ static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple) } /* - * Write relation attributes metadata to the stream. + * Write relation attributes to the stream. */ static void logicalrep_write_attrs(StringInfo out, Relation rel) { @@ -534,7 +504,7 @@ static void logicalrep_write_attrs(StringInfo out, Relation rel) } /* - * Read relation attribute metadata from the stream. + * Read relation attribute names from the stream. */ static void logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel) { diff --git a/src/gausskernel/storage/replication/logical/worker.cpp b/src/gausskernel/storage/replication/logical/worker.cpp index 7a968e2d..6647249e 100644 --- a/src/gausskernel/storage/replication/logical/worker.cpp +++ b/src/gausskernel/storage/replication/logical/worker.cpp @@ -265,11 +265,11 @@ static void slot_store_error_callback(void *arg) } /* - * Store tuple data into slot. - * - * Incoming data can be either text or binary format. + * Store data in C string form into slot. + * This is similar to BuildTupleFromCStrings but TupleTableSlot fits our + * use better. */ -static void slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, LogicalRepTupleData *tupleData) +static void slot_store_cstrings(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, char **values) { int natts = slot->tts_tupleDescriptor->natts; int i; @@ -286,52 +286,19 @@ static void slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, Lo errcallback.previous = t_thrd.log_cxt.error_context_stack; t_thrd.log_cxt.error_context_stack = &errcallback; - /* Call the "in" function for each non-dropped, non-null attribute */ + /* Call the "in" function for each non-dropped attribute */ for (i = 0; i < natts; i++) { Form_pg_attribute att = slot->tts_tupleDescriptor->attrs[i]; int remoteattnum = rel->attrmap[i]; + Oid typinput; + Oid typioparam; - if (!att->attisdropped && remoteattnum >= 0) { - StringInfo colvalue = &tupleData->colvalues[remoteattnum]; + if (!att->attisdropped && remoteattnum >= 0 && values[remoteattnum] != NULL) { errarg.remote_attnum = remoteattnum; - if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_TEXT) { - Oid typinput; - Oid typioparam; - - getTypeInputInfo(att->atttypid, &typinput, &typioparam); - slot->tts_values[i] = OidInputFunctionCall(typinput, colvalue->data, typioparam, att->atttypmod); - slot->tts_isnull[i] = false; - } else if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_BINARY) { - Oid typreceive; - Oid typioparam; - - /* - * In some code paths we may be asked to re-parse the same - * tuple data. Reset the StringInfo's cursor so that works. - */ - colvalue->cursor = 0; - - getTypeBinaryInputInfo(att->atttypid, &typreceive, &typioparam); - slot->tts_values[i] = OidReceiveFunctionCall(typreceive, colvalue, typioparam, att->atttypmod); - - /* Trouble if it didn't eat the whole buffer */ - if (colvalue->cursor != colvalue->len) { - ereport(ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION), - errmsg("incorrect binary data format in logical replication column %d", - remoteattnum + 1))); - } - slot->tts_isnull[i] = false; - } else { - /* - * NULL value from remote. (We don't expect to see - * LOGICALREP_COLUMN_UNCHANGED here, but if we do, treat it as - * NULL.) - */ - slot->tts_values[i] = (Datum) 0; - slot->tts_isnull[i] = true; - } - /* Reset attnum for error callback */ + getTypeInputInfo(att->atttypid, &typinput, &typioparam); + slot->tts_values[i] = OidInputFunctionCall(typinput, values[remoteattnum], typioparam, att->atttypmod); + slot->tts_isnull[i] = false; errarg.remote_attnum = -1; } else { /* @@ -351,19 +318,18 @@ static void slot_store_data(TupleTableSlot *slot, LogicalRepRelMapEntry *rel, Lo } /* - * Replace updated columns with data from the LogicalRepTupleData struct. + * Replace selected columns with user data provided as C strings. * This is somewhat similar to heap_modify_tuple but also calls the type * input functions on the user data. - * - * "slot" is filled with a copy of the tuple in "srcslot", replacing - * columns provided in "tupleData" and leaving others as-is. - * + * "slot" is filled with a copy of the tuple in "srcslot", with + * columns selected by the "replaces" array replaced with data values + * from "values". * Caution: unreplaced pass-by-ref columns in "slot" will point into the * storage for "srcslot". This is OK for current usage, but someday we may * need to materialize "slot" at the end to make it independent of "srcslot". */ -static void slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot, LogicalRepRelMapEntry *rel, - LogicalRepTupleData *tupleData) +static void slot_modify_cstrings(TupleTableSlot *slot, TupleTableSlot *srcslot, LogicalRepRelMapEntry *rel, + char **values, const bool *replaces) { int natts = slot->tts_tupleDescriptor->natts; int i; @@ -398,47 +364,23 @@ static void slot_modify_data(TupleTableSlot *slot, TupleTableSlot *srcslot, Logi Form_pg_attribute att = slot->tts_tupleDescriptor->attrs[i]; int remoteattnum = rel->attrmap[i]; - if (remoteattnum < 0) { + if (remoteattnum < 0 || !replaces[remoteattnum]) { continue; } - if (tupleData->colstatus[remoteattnum] != LOGICALREP_COLUMN_UNCHANGED) { - StringInfo colvalue = &tupleData->colvalues[remoteattnum]; + if (values[remoteattnum] != NULL) { + Oid typinput; + Oid typioparam; + errarg.remote_attnum = remoteattnum; - if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_TEXT) { - Oid typinput; - Oid typioparam; - - getTypeInputInfo(att->atttypid, &typinput, &typioparam); - slot->tts_values[i] = OidInputFunctionCall(typinput, colvalue->data, typioparam, att->atttypmod); - slot->tts_isnull[i] = false; - } else if (tupleData->colstatus[remoteattnum] == LOGICALREP_COLUMN_BINARY) { - Oid typreceive; - Oid typioparam; - - /* - * In some code paths we may be asked to re-parse the same - * tuple data. Reset the StringInfo's cursor so that works. - */ - colvalue->cursor = 0; - - getTypeBinaryInputInfo(att->atttypid, &typreceive, &typioparam); - slot->tts_values[i] = OidReceiveFunctionCall(typreceive, colvalue, typioparam, att->atttypmod); - - /* Trouble if it didn't eat the whole buffer */ - if (colvalue->cursor != colvalue->len) { - ereport(ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION), - errmsg("incorrect binary data format in logical replication column %d", remoteattnum + 1))); - } - slot->tts_isnull[i] = false; - } else { - /* must be LOGICALREP_COLUMN_NULL */ - slot->tts_values[i] = (Datum) 0; - slot->tts_isnull[i] = true; - } - + getTypeInputInfo(att->atttypid, &typinput, &typioparam); + slot->tts_values[i] = OidInputFunctionCall(typinput, values[remoteattnum], typioparam, att->atttypmod); + slot->tts_isnull[i] = false; errarg.remote_attnum = -1; + } else { + slot->tts_values[i] = (Datum)0; + slot->tts_isnull[i] = true; } } @@ -574,7 +516,7 @@ static void apply_handle_insert(StringInfo s) PushActiveSnapshot(GetTransactionSnapshot()); /* Process and store remote tuple in the slot */ oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); - slot_store_data(remoteslot, rel, &newtup); + slot_store_cstrings(remoteslot, rel, newtup.values); slot_fill_defaults(rel, estate, remoteslot); MemoryContextSwitchTo(oldctx); @@ -704,7 +646,7 @@ static void apply_handle_update(StringInfo s) int remoteattnum = rel->attrmap[i]; if (!att->attisdropped && remoteattnum >= 0) { Assert(remoteattnum < newtup.ncols); - if (newtup.colstatus[i] != LOGICALREP_COLUMN_UNCHANGED) { + if (newtup.changed[i]) { target_rte->updatedCols = bms_add_member(target_rte->updatedCols, i + 1 - FirstLowInvalidHeapAttributeNumber); } @@ -718,7 +660,7 @@ static void apply_handle_update(StringInfo s) /* Build the search tuple. */ oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); - slot_store_data(remoteslot, rel, has_oldtup ? &oldtup : &newtup); + slot_store_cstrings(remoteslot, rel, has_oldtup ? oldtup.values : newtup.values); MemoryContextSwitchTo(oldctx); /* @@ -741,7 +683,7 @@ static void apply_handle_update(StringInfo s) if (found) { /* Process and store remote tuple in the slot */ oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); - slot_modify_data(remoteslot, localslot, rel, &newtup); + slot_modify_cstrings(remoteslot, localslot, rel, newtup.values, newtup.changed); MemoryContextSwitchTo(oldctx); EvalPlanQualSetSlot(&epqstate, remoteslot); @@ -806,7 +748,7 @@ static void apply_handle_delete(StringInfo s) /* Find the tuple using the replica identity index. */ oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate)); - slot_store_data(remoteslot, rel, &oldtup); + slot_store_cstrings(remoteslot, rel, oldtup.values); MemoryContextSwitchTo(oldctx); /* @@ -1282,17 +1224,6 @@ static void reread_subscription(void) proc_exit(0); } - /* - * Exit if any parameter that affects the remote connection was changed. - * The launcher will start a new worker. - */ - if (strcmp(newsub->name, t_thrd.applyworker_cxt.mySubscription->name) != 0 || - newsub->binary != t_thrd.applyworker_cxt.mySubscription->binary) { - ereport(LOG, (errmsg("logical replication apply worker for subscription \"%s\" " - "will restart because of a parameter change", t_thrd.applyworker_cxt.mySubscription->name))); - proc_exit(0); - } - /* !slotname should never happen when enabled is true. */ Assert(newsub->slotname); @@ -1516,7 +1447,6 @@ void ApplyWorkerMain() options.slotname = t_thrd.applyworker_cxt.mySubscription->slotname; options.protoVersion = LOGICALREP_PROTO_VERSION_NUM; options.publicationNames = t_thrd.applyworker_cxt.mySubscription->publications; - options.binary = t_thrd.applyworker_cxt.mySubscription->binary; /* Start streaming from the slot. */ (WalReceiverFuncTable[GET_FUNC_IDX]).walrcv_startstreaming(&options); diff --git a/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp b/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp index 9f4fa440..425fd6d2 100644 --- a/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp +++ b/src/gausskernel/storage/replication/pgoutput/pgoutput.cpp @@ -15,8 +15,6 @@ #include "catalog/pg_publication.h" -#include "commands/defrem.h" - #include "replication/logical.h" #include "replication/logicalproto.h" #include "replication/origin.h" @@ -76,14 +74,11 @@ void _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->shutdown_cb = pgoutput_shutdown; } -static void parse_output_parameters(List* options, PGOutputData* data) +static void parse_output_parameters(List *options, PGOutputData *data) { ListCell *lc; bool protocol_version_given = false; bool publication_names_given = false; - bool binary_option_given = false; - - data->binary = false; foreach (lc, options) { DefElem *defel = (DefElem *)lfirst(lc); @@ -113,12 +108,6 @@ static void parse_output_parameters(List* options, PGOutputData* data) if (!SplitIdentifierString(strVal(defel->arg), ',', &(data->publication_names))) ereport(ERROR, (errcode(ERRCODE_INVALID_NAME), errmsg("invalid publication_names syntax"))); - } else if (strcmp(defel->defname, "binary") == 0) { - if (binary_option_given) - ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); - binary_option_given = true; - - data->binary = defGetBoolean(defel); } else elog(ERROR, "unrecognized pgoutput option: %s", defel->defname); } @@ -311,19 +300,19 @@ static void pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, switch (change->action) { case REORDER_BUFFER_CHANGE_INSERT: OutputPluginPrepareWrite(ctx, true); - logicalrep_write_insert(ctx->out, relation, &change->data.tp.newtuple->tuple, data->binary); + logicalrep_write_insert(ctx->out, relation, &change->data.tp.newtuple->tuple); OutputPluginWrite(ctx, true); break; case REORDER_BUFFER_CHANGE_UINSERT: OutputPluginPrepareWrite(ctx, true); - logicalrep_write_insert(ctx->out, relation, (HeapTuple)(&change->data.utp.newtuple->tuple), data->binary); + logicalrep_write_insert(ctx->out, relation, (HeapTuple)(&change->data.utp.newtuple->tuple)); OutputPluginWrite(ctx, true); break; case REORDER_BUFFER_CHANGE_UPDATE: { HeapTuple oldtuple = change->data.tp.oldtuple ? &change->data.tp.oldtuple->tuple : NULL; OutputPluginPrepareWrite(ctx, true); - logicalrep_write_update(ctx->out, relation, oldtuple, &change->data.tp.newtuple->tuple, data->binary); + logicalrep_write_update(ctx->out, relation, oldtuple, &change->data.tp.newtuple->tuple); OutputPluginWrite(ctx, true); break; } @@ -331,14 +320,14 @@ static void pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, HeapTuple oldtuple = change->data.utp.oldtuple ? ((HeapTuple)(&change->data.utp.oldtuple->tuple)) : NULL; OutputPluginPrepareWrite(ctx, true); - logicalrep_write_update(ctx->out, relation, oldtuple, (HeapTuple)(&change->data.utp.newtuple->tuple), data->binary); + logicalrep_write_update(ctx->out, relation, oldtuple, (HeapTuple)(&change->data.utp.newtuple->tuple)); OutputPluginWrite(ctx, true); break; } case REORDER_BUFFER_CHANGE_DELETE: if (change->data.tp.oldtuple) { OutputPluginPrepareWrite(ctx, true); - logicalrep_write_delete(ctx->out, relation, &change->data.tp.oldtuple->tuple, data->binary); + logicalrep_write_delete(ctx->out, relation, &change->data.tp.oldtuple->tuple); OutputPluginWrite(ctx, true); } else elog(DEBUG1, "didn't send DELETE change because of missing oldtuple"); @@ -346,7 +335,7 @@ static void pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, case REORDER_BUFFER_CHANGE_UDELETE: if (change->data.utp.oldtuple) { OutputPluginPrepareWrite(ctx, true); - logicalrep_write_delete(ctx->out, relation, (HeapTuple)(&change->data.utp.oldtuple->tuple), data->binary); + logicalrep_write_delete(ctx->out, relation, (HeapTuple)(&change->data.utp.oldtuple->tuple)); OutputPluginWrite(ctx, true); } else elog(DEBUG1, "didn't send DELETE change because of missing oldtuple"); diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index 2b7ecd05..12d1c3a5 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -50,15 +50,13 @@ CATALOG(pg_subscription,6126) BKI_SHARED_RELATION BKI_ROWTYPE_OID(6128) BKI_SCHE NameData subslotname; /* Slot name on publisher */ text subsynccommit; /* Synchronous commit setting for worker */ text subpublications[1]; /* List of publications subscribed to */ - bool subbinary; /* True if the subscription wants the - * publisher to send data in binary */ #endif } FormData_pg_subscription; typedef FormData_pg_subscription *Form_pg_subscription; -#define Natts_pg_subscription 9 +#define Natts_pg_subscription 8 #define Anum_pg_subscription_subdbid 1 #define Anum_pg_subscription_subname 2 #define Anum_pg_subscription_subowner 3 @@ -67,7 +65,6 @@ typedef FormData_pg_subscription *Form_pg_subscription; #define Anum_pg_subscription_subslotname 6 #define Anum_pg_subscription_subsynccommit 7 #define Anum_pg_subscription_subpublications 8 -#define Anum_pg_subscription_subbinary 9 typedef struct Subscription { @@ -80,7 +77,6 @@ typedef struct Subscription { char *slotname; /* Name of the replication slot */ char *synccommit; /* Synchronous commit setting for worker */ List *publications; /* List of publication names to subscribe to */ - bool binary; /* Indicates if the subscription wants data in binary format */ } Subscription; diff --git a/src/include/miscadmin.h b/src/include/miscadmin.h index 80e2ab6d..b3f8660b 100644 --- a/src/include/miscadmin.h +++ b/src/include/miscadmin.h @@ -90,7 +90,6 @@ extern const uint32 SUPPORT_DATA_REPAIR; extern const uint32 SCAN_BATCH_MODE_VERSION_NUM; extern const uint32 RELMAP_4K_VERSION_NUM; extern const uint32 PUBLICATION_VERSION_NUM; -extern const uint32 SUBSCRIPTION_BINARY_VERSION_NUM; extern const uint32 ANALYZER_HOOK_VERSION_NUM; extern const uint32 SUPPORT_HASH_XLOG_VERSION_NUM; extern const uint32 PITR_INIT_VERSION_NUM; diff --git a/src/include/replication/libpqwalreceiver.h b/src/include/replication/libpqwalreceiver.h index 96e15fc9..ba0ce4c2 100755 --- a/src/include/replication/libpqwalreceiver.h +++ b/src/include/replication/libpqwalreceiver.h @@ -35,7 +35,6 @@ typedef struct LibpqrcvConnectParam { bool logical; uint32 protoVersion; /* Logical protocol version */ List *publicationNames; /* String list of publications */ - bool binary; /* Ask publisher to use binary */ }LibpqrcvConnectParam; extern int32 pg_atoi(char* s, int size, int c); diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index 5ca74b1b..42fb59f1 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -35,21 +35,12 @@ * Keep in mind that the columns correspond to the *remote* table. */ typedef struct LogicalRepTupleData { - /* Array of StringInfos, one per column; some may be unused */ - StringInfoData *colvalues; - /* Array of markers for null/unchanged/text/binary, one per column */ - char *colstatus; + char *values[MaxTupleAttributeNumber]; /* value in out function format or NULL if values is NULL */ + bool changed[MaxTupleAttributeNumber]; /* marker for changed/unchanged values */ /* Length of above arrays */ int ncols; } LogicalRepTupleData; -/* Possible values for LogicalRepTupleData.colstatus[colnum] */ -/* These values are also used in the on-the-wire protocol */ -#define LOGICALREP_COLUMN_NULL 'n' -#define LOGICALREP_COLUMN_UNCHANGED 'u' -#define LOGICALREP_COLUMN_TEXT 't' -#define LOGICALREP_COLUMN_BINARY 'b' /* added in PG14 */ - typedef uint32 LogicalRepRelId; /* Relation information */ @@ -90,12 +81,12 @@ extern void logicalrep_read_begin(StringInfo in, LogicalRepBeginData *begin_data extern void logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn); extern void logicalrep_read_commit(StringInfo in, LogicalRepCommitData *commit_data); extern void logicalrep_write_origin(StringInfo out, const char *origin, XLogRecPtr origin_lsn); -extern void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple, bool binary); +extern void logicalrep_write_insert(StringInfo out, Relation rel, HeapTuple newtuple); extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup); -extern void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple, bool binary); +extern void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple, HeapTuple newtuple); extern LogicalRepRelId logicalrep_read_update(StringInfo in, bool *has_oldtuple, LogicalRepTupleData *oldtup, LogicalRepTupleData *newtup); -extern void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple, bool binary); +extern void logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple); extern LogicalRepRelId logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup); extern void logicalrep_write_rel(StringInfo out, Relation rel); extern LogicalRepRelation *logicalrep_read_rel(StringInfo in); diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h index 0a2fd357..fb63f49b 100644 --- a/src/include/replication/pgoutput.h +++ b/src/include/replication/pgoutput.h @@ -24,7 +24,6 @@ typedef struct PGOutputData { List *publication_names; List *publications; - bool binary; } PGOutputData; #endif /* PGOUTPUT_H */ diff --git a/src/test/regress/input/subscription.source b/src/test/regress/input/subscription.source index f47d0a50..ff384b61 100644 --- a/src/test/regress/input/subscription.source +++ b/src/test/regress/input/subscription.source @@ -37,7 +37,7 @@ CREATE SUBSCRIPTION testsub_maskconninfo CONNECTION 'host=''1.2.3.4'' port=''123 ALTER SUBSCRIPTION testsub CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234'''; ALTER SUBSCRIPTION testsub CONNECTION 'dbname=does_not_exist'; reset client_min_messages; -select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications, subbinary from pg_subscription where subname='testsub'; +select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications from pg_subscription where subname='testsub'; --- alter subscription ------ set publication ALTER SUBSCRIPTION testsub SET PUBLICATION testpub2, testpub3; @@ -57,9 +57,6 @@ select subname, subenabled, subsynccommit from pg_subscription where subname='t ALTER SUBSCRIPTION testsub SET (slot_name='testsub'); -- alter owner ALTER SUBSCRIPTION testsub owner to regress_subscription_user2; --- alter subbinary to true -ALTER SUBSCRIPTION testsub SET (binary=true); -select subname, subbinary from pg_subscription where subname='testsub'; --rename ALTER SUBSCRIPTION testsub rename to testsub_rename; --- inside a transaction block diff --git a/src/test/regress/output/subscription.source b/src/test/regress/output/subscription.source index 11145e89..d2317de8 100644 --- a/src/test/regress/output/subscription.source +++ b/src/test/regress/output/subscription.source @@ -76,10 +76,10 @@ CREATE SUBSCRIPTION testsub_maskconninfo CONNECTION 'host=''1.2.3.4'' port=''123 ALTER SUBSCRIPTION testsub CONNECTION 'host=''1.2.3.4'' port=''12345'' user=''username'' dbname=''postgres'' password=''password_1234'''; ALTER SUBSCRIPTION testsub CONNECTION 'dbname=does_not_exist'; reset client_min_messages; -select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications, subbinary from pg_subscription where subname='testsub'; - subname | owner | subenabled | subconninfo | subpublications | subbinary ----------+---------------------------+------------+----------------------+-----------------+----------- - testsub | regress_subscription_user | f | dbname=doesnotexist | {testpub} | f +select subname, pg_get_userbyid(subowner) as Owner, subenabled, subconninfo, subpublications from pg_subscription where subname='testsub'; + subname | owner | subenabled | subconninfo | subpublications +---------+---------------------------+------------+------------------------+----------------- + testsub | regress_subscription_user | f | dbname=does_not_exist | {testpub} (1 row) --- alter subscription @@ -122,14 +122,6 @@ ALTER SUBSCRIPTION testsub SET (slot_name='testsub'); ERROR: Currently enabled=false, cannot change slot_name to a non-null value. -- alter owner ALTER SUBSCRIPTION testsub owner to regress_subscription_user2; --- alter subbinary to true -ALTER SUBSCRIPTION testsub SET (binary=true); -select subname, subbinary from pg_subscription where subname='testsub'; - subname | subbinary ----------+----------- - testsub | t -(1 row) - --rename ALTER SUBSCRIPTION testsub rename to testsub_rename; --- inside a transaction block @@ -289,11 +281,10 @@ SELECT object_name,detail_info FROM pg_query_audit('2022-01-13 9:30:00', '2031-1 testsub_maskconninfo | ALTER SUBSCRIPTION testsub_maskconninfo SET (conninfo='*************************************************************************************************************************; testsub | ALTER SUBSCRIPTION testsub SET (synchronous_commit=on); testsub | ALTER SUBSCRIPTION testsub owner to regress_subscription_user2; - testsub | ALTER SUBSCRIPTION testsub SET (binary=true); testsub | ALTER SUBSCRIPTION testsub rename to testsub_rename; testsub_rename | DROP SUBSCRIPTION IF EXISTS testsub_rename; testsub_maskconninfo | DROP SUBSCRIPTION IF EXISTS testsub_maskconninfo; -(15 rows) +(14 rows) --clear audit log SELECT pg_delete_audit('1012-11-10', '3012-11-11');