Check schema owner in task_tracker_assign_task

src/backend/distributed/master/master_metadata_utility.c

@@ -1346,6 +1346,21 @@ EnsureTableOwner(Oid relationId)
 }
 
 
+/*
+ * Check that the current user has owner rights to the schema, error out if
+ * not. Superusers are regarded as owners.
+ */
+void
+EnsureSchemaOwner(Oid schemaId)
+{
+	if (!pg_namespace_ownercheck(schemaId, GetUserId()))
+	{
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACLCHECK_OBJECT_TABLE,
+					   get_namespace_name(schemaId));
+	}
+}
+
+
 /*
  * Check that the current user has owner rights to sequenceRelationId, error out if
  * not. Superusers are regarded as owners.

src/backend/distributed/worker/task_tracker_protocol.c

@@ -19,7 +19,10 @@
 
 #include <time.h>
 
+#include "access/htup_details.h"
 #include "access/xact.h"
+#include "catalog/pg_namespace.h"
+#include "catalog/namespace.h"
 #include "commands/dbcommands.h"
 #include "commands/schemacmds.h"
 #include "commands/trigger.h"
@@ -33,6 +36,8 @@
 #include "storage/lwlock.h"
 #include "storage/pmsignal.h"
 #include "utils/builtins.h"
+#include "utils/syscache.h"
+#include "utils/lsyscache.h"
 
 
 /* Local functions forward declarations */
@@ -105,6 +110,10 @@ task_tracker_assign_task(PG_FUNCTION_ARGS)
 	}
 	else
 	{
+		Oid schemaId = get_namespace_oid(jobSchemaName->data, false);
+
+		EnsureSchemaOwner(schemaId);
+
 		UnlockJobResource(jobId, AccessExclusiveLock);
 	}
 
@@ -179,6 +188,7 @@ task_tracker_cleanup_job(PG_FUNCTION_ARGS)
 {
 	uint64 jobId = PG_GETARG_INT64(0);
 
+	bool schemaExists = false;
 	HASH_SEQ_STATUS status;
 	WorkerTask *currentTask = NULL;
 	StringInfo jobDirectoryName = NULL;
@@ -186,6 +196,22 @@ task_tracker_cleanup_job(PG_FUNCTION_ARGS)
 
 	CheckCitusVersion(ERROR);
 
+	jobSchemaName = JobSchemaName(jobId);
+
+	/*
+	 * We'll keep this lock for a while, but that's ok because nothing
+	 * else should be happening on this job.
+	 */
+	LockJobResource(jobId, AccessExclusiveLock);
+
+	schemaExists = JobSchemaExists(jobSchemaName);
+	if (schemaExists)
+	{
+		Oid schemaId = get_namespace_oid(jobSchemaName->data, false);
+
+		EnsureSchemaOwner(schemaId);
+	}
+
 	/*
 	 * We first clean up any open connections, and remove tasks belonging to
 	 * this job from the shared hash.
@@ -216,8 +242,6 @@ task_tracker_cleanup_job(PG_FUNCTION_ARGS)
 	jobDirectoryName = JobDirectoryName(jobId);
 	CitusRemoveDirectory(jobDirectoryName);
 
-	LockJobResource(jobId, AccessExclusiveLock);
-	jobSchemaName = JobSchemaName(jobId);
 	RemoveJobSchema(jobSchemaName);
 	UnlockJobResource(jobId, AccessExclusiveLock);
 

src/include/distributed/master_metadata_utility.h

@@ -155,6 +155,7 @@ extern void CreateTruncateTrigger(Oid relationId);
 extern char * TableOwner(Oid relationId);
 extern void EnsureTablePermissions(Oid relationId, AclMode mode);
 extern void EnsureTableOwner(Oid relationId);
+extern void EnsureSchemaOwner(Oid schemaId);
 extern void EnsureSequenceOwner(Oid sequenceOid);
 extern void EnsureSuperUser(void);
 extern void EnsureReplicationSettings(Oid relationId, char replicationModel);