BABEL: Skip locking the tuple for ENR-related tuple modifications.

Commit 09f0820095ea8b7c6ca4269454a94695c8421628 introduced
locking for individual tuples during certain operations. However,
Babelfish introduced the concept of ENR-only relations, which
store all catalog tuples in their own local cache and not in the
physical catalogs. As these relations and their catalog tuples
are completely session-local, there is no need to acquire locks on
tuples for these relations (and it would just lead to a crash
anyways since the tuples do not have an underlying TID to lock against).

Signed-off-by: Jason Teng <[email protected]>
(cherry picked from commit 9bcb02d7b5917b5ad138008beacecbfb321ba063)
(cherry picked from commit 345c2ff)

src/backend/catalog/aclchk.c

@@ -68,6 +68,7 @@
 #include "nodes/makefuncs.h"
 #include "parser/parse_func.h"
 #include "parser/parse_type.h"
+#include "parser/parser.h"
 #include "storage/lmgr.h"
 #include "utils/acl.h"
 #include "utils/aclchk_internal.h"
@@ -1779,8 +1780,12 @@ ExecGrant_Relation(InternalGrant *istmt)
 		Oid			ownerId;
 		HeapTuple	tuple;
 		ListCell   *cell_colprivs;
+		bool		is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, relOid, ENR_TSQL_TEMP));
 
-		tuple = SearchSysCacheLocked1(RELOID, ObjectIdGetDatum(relOid));
+		if (is_enr)
+			tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relOid));
+		else
+			tuple = SearchSysCacheLocked1(RELOID, ObjectIdGetDatum(relOid));
 		if (!HeapTupleIsValid(tuple))
 			elog(ERROR, "cache lookup failed for relation %u", relOid);
 		pg_class_tuple = (Form_pg_class) GETSTRUCT(tuple);
@@ -1996,7 +2001,8 @@ ExecGrant_Relation(InternalGrant *istmt)
 										 values, nulls, replaces);
 
 			CatalogTupleUpdate(relation, &newtuple->t_self, newtuple);
-			UnlockTuple(relation, &tuple->t_self, InplaceUpdateTupleLock);
+			if (!is_enr)
+				UnlockTuple(relation, &tuple->t_self, InplaceUpdateTupleLock);
 
 			/* Update initial privileges for extensions */
 			recordExtensionInitPriv(relOid, RelationRelationId, 0, new_acl);
@@ -2009,7 +2015,7 @@ ExecGrant_Relation(InternalGrant *istmt)
 
 			pfree(new_acl);
 		}
-		else
+		else if (!is_enr)
 			UnlockTuple(relation, &tuple->t_self, InplaceUpdateTupleLock);
 
 		/*
@@ -2119,8 +2125,12 @@ ExecGrant_Database(InternalGrant *istmt)
 		Oid		   *oldmembers;
 		Oid		   *newmembers;
 		HeapTuple	tuple;
+		bool		is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, datId, ENR_TSQL_TEMP));
 
-		tuple = SearchSysCacheLocked1(DATABASEOID, ObjectIdGetDatum(datId));
+		if (is_enr)
+			tuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(datId));
+		else
+			tuple = SearchSysCacheLocked1(DATABASEOID, ObjectIdGetDatum(datId));
 		if (!HeapTupleIsValid(tuple))
 			elog(ERROR, "cache lookup failed for database %u", datId);
 
@@ -2189,7 +2199,8 @@ ExecGrant_Database(InternalGrant *istmt)
 									 nulls, replaces);
 
 		CatalogTupleUpdate(relation, &newtuple->t_self, newtuple);
-		UnlockTuple(relation, &tuple->t_self, InplaceUpdateTupleLock);
+		if (!is_enr)
+			UnlockTuple(relation, &tuple->t_self, InplaceUpdateTupleLock);
 
 		/* Update the shared dependency ACL info */
 		updateAclDependencies(DatabaseRelationId, pg_database_tuple->oid, 0,

src/backend/commands/dbcommands.c

@@ -49,6 +49,7 @@
 #include "common/file_perm.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
+#include "parser/parser.h"
 #include "pgstat.h"
 #include "postmaster/bgwriter.h"
 #include "replication/slot.h"
@@ -1058,6 +1059,7 @@ RenameDatabase(const char *oldname, const char *newname)
 	int			notherbackends;
 	int			npreparedxacts;
 	ObjectAddress address;
+	bool		is_enr = false;
 
 	/*
 	 * Look up the target database's OID, and get exclusive lock on it. We
@@ -1125,13 +1127,18 @@ RenameDatabase(const char *oldname, const char *newname)
 				 errdetail_busy_db(notherbackends, npreparedxacts)));
 
 	/* rename */
-	newtup = SearchSysCacheLockedCopy1(DATABASEOID, ObjectIdGetDatum(db_id));
+	is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, db_id, ENR_TSQL_TEMP));
+	if (is_enr)
+		newtup = SearchSysCacheCopy1(DATABASEOID, ObjectIdGetDatum(db_id));
+	else
+		newtup = SearchSysCacheLockedCopy1(DATABASEOID, ObjectIdGetDatum(db_id));
 	if (!HeapTupleIsValid(newtup))
 		elog(ERROR, "cache lookup failed for database %u", db_id);
 	otid = newtup->t_self;
 	namestrcpy(&(((Form_pg_database) GETSTRUCT(newtup))->datname), newname);
 	CatalogTupleUpdate(rel, &otid, newtup);
-	UnlockTuple(rel, &otid, InplaceUpdateTupleLock);
+	if (!is_enr)
+		UnlockTuple(rel, &otid, InplaceUpdateTupleLock);
 
 	InvokeObjectPostAlterHook(DatabaseRelationId, db_id, 0);
 

src/backend/commands/indexcmds.c

@@ -48,6 +48,7 @@
 #include "parser/parse_coerce.h"
 #include "parser/parse_func.h"
 #include "parser/parse_oper.h"
+#include "parser/parser.h"
 #include "partitioning/partdesc.h"
 #include "pgstat.h"
 #include "rewrite/rewriteManip.h"
@@ -63,6 +64,7 @@
 #include "utils/memutils.h"
 #include "utils/partcache.h"
 #include "utils/pg_rusage.h"
+#include "utils/queryenvironment.h"
 #include "utils/regproc.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
@@ -4307,16 +4309,21 @@ update_relispartition(Oid relationId, bool newval)
 	HeapTuple	tup;
 	Relation	classRel;
 	ItemPointerData otid;
+	bool		is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, relationId, ENR_TSQL_TEMP));
 
 	classRel = table_open(RelationRelationId, RowExclusiveLock);
-	tup = SearchSysCacheLockedCopy1(RELOID, ObjectIdGetDatum(relationId));
+	if (is_enr)
+		tup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(relationId));
+	else
+		tup = SearchSysCacheLockedCopy1(RELOID, ObjectIdGetDatum(relationId));
 	if (!HeapTupleIsValid(tup))
 		elog(ERROR, "cache lookup failed for relation %u", relationId);
 	otid = tup->t_self;
 	Assert(((Form_pg_class) GETSTRUCT(tup))->relispartition != newval);
 	((Form_pg_class) GETSTRUCT(tup))->relispartition = newval;
 	CatalogTupleUpdate(classRel, &otid, tup);
-	UnlockTuple(classRel, &otid, InplaceUpdateTupleLock);
+	if (!is_enr)
+		UnlockTuple(classRel, &otid, InplaceUpdateTupleLock);
 	heap_freetuple(tup);
 	table_close(classRel, RowExclusiveLock);
 }

src/backend/commands/tablecmds.c

@@ -3388,13 +3388,17 @@ SetRelationTableSpace(Relation rel,
 	ItemPointerData otid;
 	Form_pg_class rd_rel;
 	Oid			reloid = RelationGetRelid(rel);
+	bool		is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, reloid, ENR_TSQL_TEMP));
 
 	Assert(CheckRelationTableSpaceMove(rel, newTableSpaceId));
 
 	/* Get a modifiable copy of the relation's pg_class row. */
 	pg_class = table_open(RelationRelationId, RowExclusiveLock);
 
-	tuple = SearchSysCacheLockedCopy1(RELOID, ObjectIdGetDatum(reloid));
+	if (is_enr)
+			tuple = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(reloid));
+	else
+		tuple = SearchSysCacheLockedCopy1(RELOID, ObjectIdGetDatum(reloid));
 	if (!HeapTupleIsValid(tuple))
 		elog(ERROR, "cache lookup failed for relation %u", reloid);
 	otid = tuple->t_self;
@@ -3406,7 +3410,8 @@ SetRelationTableSpace(Relation rel,
 	if (OidIsValid(newRelFileNode))
 		rd_rel->relfilenode = newRelFileNode;
 	CatalogTupleUpdate(pg_class, &otid, tuple);
-	UnlockTuple(pg_class, &otid, InplaceUpdateTupleLock);
+	if (!is_enr)
+		UnlockTuple(pg_class, &otid, InplaceUpdateTupleLock);
 
 	/*
 	 * Record dependency on tablespace.  This is only required for relations
@@ -3904,6 +3909,7 @@ RenameRelationInternal(Oid myrelid, const char *newrelname, bool is_internal, bo
 	HeapTuple	reltup;
 	Form_pg_class relform;
 	Oid			namespaceId;
+	bool		is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, myrelid, ENR_TSQL_TEMP));
 
 	/*
 	 * Grab a lock on the target relation, which we will NOT release until end
@@ -3923,7 +3929,10 @@ RenameRelationInternal(Oid myrelid, const char *newrelname, bool is_internal, bo
 	 */
 	relrelation = table_open(RelationRelationId, RowExclusiveLock);
 
-	reltup = SearchSysCacheLockedCopy1(RELOID, ObjectIdGetDatum(myrelid));
+	if (is_enr)
+		reltup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(myrelid));
+	else
+		reltup = SearchSysCacheLockedCopy1(RELOID, ObjectIdGetDatum(myrelid));
 	if (!HeapTupleIsValid(reltup))	/* shouldn't happen */
 		elog(ERROR, "cache lookup failed for relation %u", myrelid);
 	otid = reltup->t_self;
@@ -3952,7 +3961,8 @@ RenameRelationInternal(Oid myrelid, const char *newrelname, bool is_internal, bo
 	namestrcpy(&(relform->relname), newrelname);
 
 	CatalogTupleUpdate(relrelation, &otid, reltup);
-	UnlockTuple(relrelation, &otid, InplaceUpdateTupleLock);
+	if (!is_enr)
+		UnlockTuple(relrelation, &otid, InplaceUpdateTupleLock);
 
 	InvokeObjectPostAlterHookArg(RelationRelationId, myrelid, 0,
 								 InvalidOid, is_internal);
@@ -14012,6 +14022,7 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 	bool		repl_null[Natts_pg_class];
 	bool		repl_repl[Natts_pg_class];
 	static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
+	bool		is_enr = false;
 
 	if (defList == NIL && operation != AT_ReplaceRelOptions)
 		return;					/* nothing to do */
@@ -14020,7 +14031,11 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 
 	/* Fetch heap tuple */
 	relid = RelationGetRelid(rel);
-	tuple = SearchSysCacheLocked1(RELOID, ObjectIdGetDatum(relid));
+	is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, relid, ENR_TSQL_TEMP));
+	if (is_enr)
+		tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
+	else
+		tuple = SearchSysCacheLocked1(RELOID, ObjectIdGetDatum(relid));
 	if (!HeapTupleIsValid(tuple))
 		elog(ERROR, "cache lookup failed for relation %u", relid);
 
@@ -14123,7 +14138,8 @@ ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
 								 repl_val, repl_null, repl_repl);
 
 	CatalogTupleUpdate(pgclass, &newtuple->t_self, newtuple);
-	UnlockTuple(pgclass, &tuple->t_self, InplaceUpdateTupleLock);
+	if (!is_enr)
+		UnlockTuple(pgclass, &tuple->t_self, InplaceUpdateTupleLock);
 
 	InvokeObjectPostAlterHook(RelationRelationId, RelationGetRelid(rel), 0);
 
@@ -16320,9 +16336,13 @@ AlterRelationNamespaceInternal(Relation classRel, Oid relOid,
 	Form_pg_class classForm;
 	ObjectAddress thisobj;
 	bool		already_done = false;
+	bool		is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, relOid, ENR_TSQL_TEMP));
 
 	/* no rel lock for relkind=c so use LOCKTAG_TUPLE */
-	classTup = SearchSysCacheLockedCopy1(RELOID, ObjectIdGetDatum(relOid));
+	if (is_enr)
+		classTup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(relOid));
+	else
+		classTup = SearchSysCacheLockedCopy1(RELOID, ObjectIdGetDatum(relOid));
 	if (!HeapTupleIsValid(classTup))
 		elog(ERROR, "cache lookup failed for relation %u", relOid);
 	classForm = (Form_pg_class) GETSTRUCT(classTup);
@@ -16356,7 +16376,8 @@ AlterRelationNamespaceInternal(Relation classRel, Oid relOid,
 		classForm->relnamespace = newNspOid;
 
 		CatalogTupleUpdate(classRel, &otid, classTup);
-		UnlockTuple(classRel, &otid, InplaceUpdateTupleLock);
+		if (!is_enr)
+			UnlockTuple(classRel, &otid, InplaceUpdateTupleLock);
 
 
 		/* Update dependency on schema if caller said so */
@@ -16369,7 +16390,7 @@ AlterRelationNamespaceInternal(Relation classRel, Oid relOid,
 			elog(ERROR, "failed to change schema dependency for relation \"%s\"",
 				 NameStr(classForm->relname));
 	}
-	else
+	else if (!is_enr)
 		UnlockTuple(classRel, &classTup->t_self, InplaceUpdateTupleLock);
 	if (!already_done)
 	{

src/backend/utils/cache/relcache.c

@@ -71,6 +71,8 @@
 #include "nodes/makefuncs.h"
 #include "nodes/nodeFuncs.h"
 #include "optimizer/optimizer.h"
+#include "parser/parser.h"
+#include "postmaster/autovacuum.h"
 #include "rewrite/rewriteDefine.h"
 #include "rewrite/rowsecurity.h"
 #include "storage/lmgr.h"
@@ -3700,6 +3702,7 @@ RelationSetNewRelfilenode(Relation relation, char persistence)
 	MultiXactId minmulti = InvalidMultiXactId;
 	TransactionId freezeXid = InvalidTransactionId;
 	RelFileNode newrnode;
+	bool			is_enr = false;
 
 	/* Allocate a new relfilenode */
 	newrelfilenode = GetNewRelFileNode(relation->rd_rel->reltablespace, NULL,
@@ -3710,8 +3713,13 @@ RelationSetNewRelfilenode(Relation relation, char persistence)
 	 */
 	pg_class = table_open(RelationRelationId, RowExclusiveLock);
 
-	tuple = SearchSysCacheLockedCopy1(RELOID,
-									  ObjectIdGetDatum(RelationGetRelid(relation)));
+	is_enr = (sql_dialect == SQL_DIALECT_TSQL && get_ENR_withoid(currentQueryEnv, RelationGetRelid(relation), ENR_TSQL_TEMP));
+	if (is_enr)
+		tuple = SearchSysCacheCopy1(RELOID,
+									ObjectIdGetDatum(RelationGetRelid(relation)));
+	else
+		tuple = SearchSysCacheLockedCopy1(RELOID,
+										  ObjectIdGetDatum(RelationGetRelid(relation)));
 	if (!HeapTupleIsValid(tuple))
 		elog(ERROR, "could not find tuple for relation %u",
 			 RelationGetRelid(relation));
@@ -3817,7 +3825,8 @@ RelationSetNewRelfilenode(Relation relation, char persistence)
 		CatalogTupleUpdate(pg_class, &otid, tuple);
 	}
 
-	UnlockTuple(pg_class, &otid, InplaceUpdateTupleLock);
+	if (!is_enr)
+		UnlockTuple(pg_class, &otid, InplaceUpdateTupleLock);
 	heap_freetuple(tuple);
 
 	table_close(pg_class, RowExclusiveLock);

src/backend/utils/cache/syscache.c

@@ -1215,6 +1215,12 @@ ReleaseSysCache(HeapTuple tuple)
  *
  * The returned tuple may be the subject of an uncommitted update, so this
  * doesn't prevent the "tuple concurrently updated" error.
+ * 
+ * Note: For Babelfish, this function should not be used if the target tuple is
+ * for an ENR entry, as there is no physical tid for ENR catalog tuples (since ENR
+ * entries hold all catalog data internally in their cache). Use SearchSysCache1() instead
+ * to look up tuples for catalogs for ENR entries, and also skip the UnlockTuple() call
+ * in such cases.
  */
 HeapTuple
 SearchSysCacheLocked1(int cacheId,
@@ -1328,6 +1334,9 @@ SearchSysCacheCopy(int cacheId,
  * Meld SearchSysCacheLockedCopy1 with SearchSysCacheCopy().  After the
  * caller's heap_update(), it should UnlockTuple(InplaceUpdateTupleLock) and
  * heap_freetuple().
+ * 
+ * See SearchSysCacheLocked1() for notes about ENR entries (do not use this 
+ * function for tuples related to ENR entries).
  */
 HeapTuple
 SearchSysCacheLockedCopy1(int cacheId,