grant on schema does not take effect on future objects created in a s…

…chema by schema's authorized users

contrib/babelfishpg_tsql/src/pl_exec-2.c

@@ -3759,7 +3759,7 @@ exec_stmt_grantschema(PLtsql_execstate *estate, PLtsql_stmt_grantschema *stmt)
 		for (i = 0; i < NUMBER_OF_PERMISSIONS; i++)
 		{
 			if (stmt->privileges & permissions[i])
-				exec_grantschema_subcmds(schema_name, rolname, stmt->is_grant, stmt->with_grant_option, permissions[i]);
+				exec_grantschema_subcmds(schema_name, rolname, stmt->is_grant, stmt->with_grant_option, permissions[i], false);
 		}
 
 		if (stmt->is_grant)

contrib/babelfishpg_tsql/src/pl_handler.c

@@ -3548,7 +3548,7 @@ bbf_ProcessUtility(PlannedStmt *pstmt,
 						for (i = 0; i < NUMBER_OF_PERMISSIONS; i++)
 						{
 							/* Execute the GRANT SCHEMA subcommands. */
-							exec_grantschema_subcmds(create_schema->schemaname, rolspec->rolename, true, false, permissions[i]);
+							exec_grantschema_subcmds(create_schema->schemaname, rolspec->rolename, true, false, permissions[i], true);
 						}
 					}
 					return;

contrib/babelfishpg_tsql/src/pltsql.h

@@ -2002,6 +2002,7 @@ extern int	get_insert_bulk_kilobytes_per_batch(void);
 extern char *get_original_query_string(void);
 extern AclMode string_to_privilege(const char *privname);
 extern const char *privilege_to_string(AclMode privilege);
+extern int get_owner_of_schema(const char *schema);
 
 /*
  * Functions for namespace handling in pl_funcs.c
@@ -2064,7 +2065,7 @@ extern char *get_fulltext_index_name(Oid relid, const char *table_name);
 extern const char *gen_schema_name_for_fulltext_index(const char *schema_name);
 extern bool check_fulltext_exist(const char *schema_name, const char *table_name);
 extern bool is_unique_index(Oid relid, const char *index_name);
-extern void exec_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant, bool with_grant_option, AclMode privilege);
+extern void exec_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant, bool with_grant_option, AclMode privilege, bool is_create_schema);
 extern int	TsqlUTF8LengthInUTF16(const void *vin, int len);
 extern void TsqlCheckUTF16Length_bpchar(const char *s, int32 len, int32 maxlen, int charlen, bool isExplicit);
 extern void TsqlCheckUTF16Length_varchar(const char *s, int32 len, int32 maxlen, bool isExplicit);

contrib/babelfishpg_tsql/src/pltsql_utils.c

@@ -1,6 +1,7 @@
 #include "postgres.h"
 
 #include "catalog/namespace.h"
+#include "catalog/pg_namespace.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_type.h"
 #include "catalog/pg_trigger.h"
@@ -28,6 +29,7 @@
 #include "tcop/utility.h"
 
 #include "multidb.h"
+#include "session.h"
 
 common_utility_plugin *common_utility_plugin_ptr = NULL;
 
@@ -1108,7 +1110,7 @@ update_GrantStmt(Node *n, const char *object, const char *obj_schema, const char
 }
 
 static void
-update_AlterDefaultPrivilegesStmt(Node *n, const char *object, const char *grantee, const char *priv)
+update_AlterDefaultPrivilegesStmt(Node *n, const char *schema, const char *role, const char *grantee, const char *priv)
 {
 	AlterDefaultPrivilegesStmt *stmt = (AlterDefaultPrivilegesStmt *) n;
 	ListCell *lc;
@@ -1123,11 +1125,16 @@ update_AlterDefaultPrivilegesStmt(Node *n, const char *object, const char *grant
 
 	foreach(lc, stmt->options)
 	{
-		if (object)
+		DefElem    *defel = (DefElem *) lfirst(lc);
+
+		if (schema && defel->arg && strcmp(defel->defname, "schemas") == 0)
 		{
-			DefElem *tmp = (DefElem *) lfirst(lc);
-			tmp->defname = pstrdup("schemas");
-			tmp->arg = (Node *)list_make1(makeString((char *)object));
+			defel->arg = (Node *)list_make1(makeString((char *)schema));
+		}
+
+		else if (role && defel->arg && strcmp(defel->defname, "roles") == 0)
+		{
+			defel->arg = (Node *)list_make1(makeString((char *)role));
 		}
 	}
 }
@@ -2135,13 +2142,30 @@ exec_alter_role_cmd(char *query_str, RoleSpec *role)
  * Helper function to generate GRANT on SCHEMA subcommands.
  */
 static List
-*gen_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant, bool with_grant_option, AclMode privilege)
+*gen_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant, bool with_grant_option, AclMode privilege, bool is_create_schema)
 {
 	StringInfoData query;
 	List	   *stmt_list;
 	Node	   *stmt;
 	int			expected_stmts = 2;
 	int			i = 0;
+	bool		owner_other_than_dbo = false;
+	char 		*dbname = get_cur_db_name();
+	const char	*dbo_role = get_dbo_role_name(dbname);
+	const char	*db_owner_role = get_db_owner_name(dbname);
+	const char	*schema_owner = GetUserNameFromId(get_owner_of_schema(schema), false);
+
+	/*
+	 * Don't need multiple ALTER DEFAULT PRIVILEGE statements, if:
+	 * 1. It's a part of CREATE SCHEMA statement
+	 * 2. If the schema owner is not dbo
+	 * 3. If the schema owner is not db_owner
+	 */
+	if ((strcmp(dbo_role, schema_owner) != 0) && (strcmp(db_owner_role, schema_owner) != 0) && !is_create_schema)
+	{
+		owner_other_than_dbo = true;
+	}
+
 	initStringInfo(&query);
 	if (is_grant)
 	{
@@ -2164,7 +2188,15 @@ static List
 				appendStringInfo(&query, "GRANT dummy ON ALL TABLES IN SCHEMA dummy TO dummy WITH GRANT OPTION; ");
 			else
 				appendStringInfo(&query, "GRANT dummy ON ALL TABLES IN SCHEMA dummy TO dummy; ");
-			appendStringInfo(&query, "ALTER DEFAULT PRIVILEGES IN SCHEMA dummy GRANT dummy ON TABLES TO dummy; ");
+
+			if (owner_other_than_dbo)
+			{
+				appendStringInfo(&query, "ALTER DEFAULT PRIVILEGES FOR ROLE dummy IN SCHEMA dummy GRANT dummy ON TABLES TO dummy; ");
+				appendStringInfo(&query, "ALTER DEFAULT PRIVILEGES FOR ROLE dummy IN SCHEMA dummy GRANT dummy ON TABLES TO dummy; ");
+				expected_stmts = 3;
+			}
+			else
+				appendStringInfo(&query, "ALTER DEFAULT PRIVILEGES IN SCHEMA dummy GRANT dummy ON TABLES TO dummy; ");
 		}
 	}
 	else
@@ -2177,7 +2209,14 @@ static List
 		else
 		{
 			appendStringInfo(&query, "REVOKE dummy ON ALL TABLES IN SCHEMA dummy FROM dummy; ");
-			appendStringInfo(&query, "ALTER DEFAULT PRIVILEGES IN SCHEMA dummy REVOKE dummy ON TABLES FROM dummy; ");
+			if (owner_other_than_dbo)
+			{
+				appendStringInfo(&query, "ALTER DEFAULT PRIVILEGES FOR ROLE dummy IN SCHEMA dummy REVOKE dummy ON TABLES FROM dummy; ");
+				appendStringInfo(&query, "ALTER DEFAULT PRIVILEGES FOR ROLE dummy IN SCHEMA dummy REVOKE dummy ON TABLES FROM dummy; ");
+				expected_stmts = 3;
+			}
+			else
+				appendStringInfo(&query, "ALTER DEFAULT PRIVILEGES IN SCHEMA dummy REVOKE dummy ON TABLES FROM dummy; ");
 		}
 	}
 	stmt_list = raw_parser(query.data, RAW_PARSE_DEFAULT);
@@ -2194,7 +2233,18 @@ static List
 	if (privilege == ACL_EXECUTE)
 		update_GrantStmt(stmt, schema, NULL, rolname, privilege_to_string(privilege));
 	else
-		update_AlterDefaultPrivilegesStmt(stmt, schema, rolname, privilege_to_string(privilege));
+	{
+		if (owner_other_than_dbo)
+		{
+			update_AlterDefaultPrivilegesStmt(stmt, schema, dbo_role, rolname, privilege_to_string(privilege));
+			stmt = parsetree_nth_stmt(stmt_list, i++);
+			update_AlterDefaultPrivilegesStmt(stmt, schema, schema_owner, rolname, privilege_to_string(privilege));
+		}
+		else
+		{
+			update_AlterDefaultPrivilegesStmt(stmt, schema, NULL, rolname, privilege_to_string(privilege));
+		}
+	}
 
 	pfree(query.data);
 	return stmt_list;
@@ -2206,12 +2256,18 @@ static List
  * inputs are sanitized to prevent unexpected behaviour.
  */
 void
-exec_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant, bool with_grant_option, AclMode privilege)
+exec_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant, bool with_grant_option, AclMode privilege, bool is_create_schema)
 {
 	List		*parsetree_list;
 	ListCell	*parsetree_item;
+	const char *prev_current_user;
+	const char *dbo_role = get_dbo_role_name(get_cur_db_name());
+
+	/* Need dbo user to execute the statements. */
+	prev_current_user = GetUserNameFromId(GetUserId(), false);
+	bbf_set_current_user(dbo_role);
 
-	parsetree_list = gen_grantschema_subcmds(schema, rolname, is_grant, with_grant_option, privilege);
+	parsetree_list = gen_grantschema_subcmds(schema, rolname, is_grant, with_grant_option, privilege, is_create_schema);
 	/* Run all subcommands */
 	foreach(parsetree_item, parsetree_list)
 	{
@@ -2236,6 +2292,7 @@ exec_grantschema_subcmds(const char *schema, const char *rolname, bool is_grant,
 					None_Receiver,
 					NULL);
 	}
+	bbf_set_current_user(prev_current_user);
 }
 
 AclMode
@@ -2306,3 +2363,30 @@ make_rolespec_node(const char *rolename)
 
 	return n;
 }
+
+/* 
+ * Returns the oid of schema owner.
+ */
+int
+get_owner_of_schema(const char *schema)
+{
+	HeapTuple	tup;
+	Form_pg_namespace nspform;
+	int result;
+	NameData   *schema_name;
+	schema_name = (NameData *) palloc0(NAMEDATALEN);
+	snprintf(schema_name->data, NAMEDATALEN, "%s", schema);
+
+	tup = SearchSysCache1(NAMESPACENAME, NameGetDatum(schema_name));
+
+	if (!HeapTupleIsValid(tup))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_SCHEMA),
+				 errmsg("schema \"%s\" does not exist", schema)));
+
+	nspform = (Form_pg_namespace) GETSTRUCT(tup);
+	result = ((int) nspform->nspowner);
+	ReleaseSysCache(tup);
+
+	return result;
+}

contrib/babelfishpg_tsql/src/schemacmds.c

@@ -34,7 +34,7 @@ add_ns_ext_info(CreateSchemaStmt *stmt, const char *queryString, const char *ori
 
 	/*
 	 * orig_name will be provided only when queryString is not valid. e.g
-	 * CREATE LOGICLA DATABASE
+	 * CREATE LOGICAL DATABASE
 	 */
 	if (!orig_name)
 	{