Fix issue #111

include/pgactive.h

@@ -478,6 +478,7 @@ extern int	pgactive_init_node_parallel_jobs;
 extern int	pgactive_max_nodes;
 extern bool pgactive_permit_node_identifier_getter_function_creation;
 extern bool pgactive_debug_trace_connection_errors;
+extern bool pgactive_apply_as_table_owner;
 
 static const char *const pgactive_default_apply_connection_options =
 "connect_timeout=30 "

src/pgactive.c

@@ -126,6 +126,7 @@ int			pgactive_init_node_parallel_jobs;
 int			pgactive_max_nodes;
 bool		pgactive_permit_node_identifier_getter_function_creation;
 bool		pgactive_debug_trace_connection_errors;
+bool		pgactive_apply_as_table_owner;
 
 PG_MODULE_MAGIC;
 
@@ -1251,6 +1252,16 @@ _PG_init(void)
 							 GUC_SUPERUSER_ONLY,
 							 NULL, NULL, NULL);
 
+	DefineCustomBoolVariable("pgactive.apply_as_table_owner",
+							 "Apply DML changes as the table owner instead of superuser.",
+							 "When enabled, the apply worker switches to the table owner "
+							 "before executing INSERT, UPDATE, or DELETE operations.",
+							 &pgactive_apply_as_table_owner,
+							 true,
+							 PGC_POSTMASTER,
+							 0,
+							 NULL, NULL, NULL);
+
 	EmitWarningsOnPlaceholders("pgactive");
 
 	/* Security label provider hook */

src/pgactive_apply.c

@@ -65,6 +65,30 @@
 #include "utils/memutils.h"
 #include "utils/snapmgr.h"
 
+#if PG_VERSION_NUM >= 160000
+#include "utils/usercontext.h"
+#else
+typedef struct UserContext
+{
+	Oid			save_userid;
+	int			save_sec_context;
+} UserContext;
+
+static inline void
+SwitchToUntrustedUser(Oid userid, UserContext *context)
+{
+	GetUserIdAndSecContext(&context->save_userid, &context->save_sec_context);
+	SetUserIdAndSecContext(userid,
+						   context->save_sec_context | SECURITY_RESTRICTED_OPERATION);
+}
+
+static inline void
+RestoreUserContext(UserContext *context)
+{
+	SetUserIdAndSecContext(context->save_userid, context->save_sec_context);
+}
+#endif
+
 /*
  * Useful for development:
  * #define VERBOSE_INSERT
@@ -606,6 +630,7 @@ process_remote_insert(StringInfo s)
 	ItemPointerData conflicting_tid;
 	ErrorContextCallback errcallback;
 	struct ActionErrCallbackArg cbarg;
+	UserContext ucxt;
 
 	ItemPointerSetInvalid(&conflicting_tid);
 
@@ -623,6 +648,14 @@ process_remote_insert(StringInfo s)
 
 	rel = read_rel(s, RowExclusiveLock, &cbarg);
 
+	if (pgactive_apply_as_table_owner)
+	{
+		SwitchToUntrustedUser(rel->rel->rd_rel->relowner, &ucxt);
+		elog(DEBUG1, "pgactive apply INSERT as user %s on %s",
+			 GetUserNameFromId(rel->rel->rd_rel->relowner, false),
+			 RelationGetRelationName(rel->rel));
+	}
+
 	emit_replay_info(&cbarg);
 
 	action = pq_getmsgbyte(s);
@@ -842,6 +875,9 @@ process_remote_insert(StringInfo s)
 
 	PopActiveSnapshot();
 
+	if (pgactive_apply_as_table_owner)
+		RestoreUserContext(&ucxt);
+
 	ExecCloseIndices(relinfo);
 
 	check_pgactive_wakeups(rel);
@@ -933,6 +969,7 @@ process_remote_update(StringInfo s)
 	ErrorContextCallback errcallback;
 	struct ActionErrCallbackArg cbarg;
 	ResultRelInfo *relinfo = makeNode(ResultRelInfo);
+	UserContext ucxt;
 
 	xact_action_counter++;
 	memset(&cbarg, 0, sizeof(struct ActionErrCallbackArg));
@@ -946,6 +983,14 @@ process_remote_update(StringInfo s)
 
 	rel = read_rel(s, RowExclusiveLock, &cbarg);
 
+	if (pgactive_apply_as_table_owner)
+	{
+		SwitchToUntrustedUser(rel->rel->rd_rel->relowner, &ucxt);
+		elog(DEBUG1, "pgactive apply UPDATE as user %s on %s",
+			 GetUserNameFromId(rel->rel->rd_rel->relowner, false),
+			 RelationGetRelationName(rel->rel));
+	}
+
 	emit_replay_info(&cbarg);
 
 	action = pq_getmsgbyte(s);
@@ -1183,6 +1228,9 @@ process_remote_update(StringInfo s)
 
 	PopActiveSnapshot();
 
+	if (pgactive_apply_as_table_owner)
+		RestoreUserContext(&ucxt);
+
 	check_pgactive_wakeups(rel);
 
 	/* release locks upon commit */
@@ -1213,6 +1261,7 @@ process_remote_delete(StringInfo s)
 	ErrorContextCallback errcallback;
 	struct ActionErrCallbackArg cbarg;
 	ResultRelInfo *relinfo = makeNode(ResultRelInfo);
+	UserContext ucxt;
 
 	Assert(pgactive_apply_worker != NULL);
 
@@ -1228,6 +1277,14 @@ process_remote_delete(StringInfo s)
 
 	rel = read_rel(s, RowExclusiveLock, &cbarg);
 
+	if (pgactive_apply_as_table_owner)
+	{
+		SwitchToUntrustedUser(rel->rel->rd_rel->relowner, &ucxt);
+		elog(DEBUG1, "pgactive apply DELETE as user %s on %s",
+			 GetUserNameFromId(rel->rel->rd_rel->relowner, false),
+			 RelationGetRelationName(rel->rel));
+	}
+
 	emit_replay_info(&cbarg);
 
 	action = pq_getmsgbyte(s);
@@ -1238,6 +1295,8 @@ process_remote_delete(StringInfo s)
 	if (action == 'E')
 	{
 		elog(WARNING, "got delete without pkey");
+		if (pgactive_apply_as_table_owner)
+			RestoreUserContext(&ucxt);
 		pgactive_table_close(rel, NoLock);
 		return;
 	}
@@ -1367,6 +1426,9 @@ process_remote_delete(StringInfo s)
 
 	PopActiveSnapshot();
 
+	if (pgactive_apply_as_table_owner)
+		RestoreUserContext(&ucxt);
+
 	check_pgactive_wakeups(rel);
 
 	index_close(idxrel, NoLock);