Fix transaction table reservation.

Author: asfernandes

src/common/classes/ClumpletReader.cpp

@@ -292,10 +292,11 @@ ClumpletReader::ClumpletType ClumpletReader::getClumpletType(UCHAR tag) const
 	case Tpb:
 		switch (tag)
 		{
-        case isc_tpb_lock_write:
-        case isc_tpb_lock_read:
+		case isc_tpb_lock_write:
+		case isc_tpb_lock_read:
 		case isc_tpb_lock_timeout:
 		case isc_tpb_at_snapshot_number:
+        case isc_tpb_lock_table_schema:
 			return TraditionalDpb;
 		}
 		return SingleTpb;

src/dsql/StmtNodes.cpp

@@ -9873,6 +9873,9 @@ void SetTransactionNode::execute(thread_db* tdbb, DsqlRequest* request, jrd_tra*
 void SetTransactionNode::genTableLock(DsqlCompilerScratch* dsqlScratch,
 	const RestrictionOption& tblLock, USHORT lockLevel)
 {
+	const auto tdbb = JRD_get_thread_data();
+	const auto attachment = dsqlScratch->getAttachment()->dbb_attachment;
+
 	if (tblLock.tables->isEmpty())
 		return;
 
@@ -9884,11 +9887,19 @@ void SetTransactionNode::genTableLock(DsqlCompilerScratch* dsqlScratch,
 	const USHORT lockMode = (tblLock.lockMode & LOCK_MODE_WRITE) ?
 		isc_tpb_lock_write : isc_tpb_lock_read;
 
-	for (const auto& tableName : *tblLock.tables)
+	for (auto& tableName : *tblLock.tables)
 	{
 		dsqlScratch->appendUChar(lockMode);
-		// FIXME: schema
-		dsqlScratch->appendNullString(tableName.object.c_str());	// stuff table name
+		dsqlScratch->appendNullString(attachment->nameToUserCharSet(tdbb, tableName.object).c_str());
+
+		dsqlScratch->qualifyExistingName(tableName, obj_relation);
+
+		if (tableName.schema.hasData())
+		{
+			dsqlScratch->appendUChar(isc_tpb_lock_table_schema);
+			dsqlScratch->appendNullString(attachment->nameToUserCharSet(tdbb, tableName.schema).c_str());
+		}
+
 		dsqlScratch->appendUChar(lockLevel);
 	}
 }

src/include/firebird/impl/consts_pub.h

@@ -274,6 +274,7 @@
 #define isc_tpb_read_consistency          22
 #define isc_tpb_at_snapshot_number        23
 #define isc_tpb_auto_release_temp_blobid  24
+#define isc_tpb_lock_table_schema         25
 
 
 /************************/

src/include/gen/Firebird.pas

@@ -4189,6 +4189,7 @@ 	IProfilerStatsImpl = class(IProfilerStats)
 	isc_tpb_read_consistency = byte(22);
 	isc_tpb_at_snapshot_number = byte(23);
 	isc_tpb_auto_release_temp_blobid = byte(24);
+	isc_tpb_lock_table_schema = byte(25);
 	isc_bpb_version1 = byte(1);
 	isc_bpb_source_type = byte(1);
 	isc_bpb_target_type = byte(2);

src/jrd/tra.cpp

@@ -3163,6 +3163,11 @@ static void transaction_options(thread_db* tdbb,
 			transaction->tra_flags |= TRA_auto_release_temp_blobid;
 			break;
 
+		case isc_tpb_lock_table_schema:
+			ERR_post(Arg::Gds(isc_bad_tpb_content) <<
+					 Arg::Gds(isc_tpb_reserv_before_table) << Arg::Str("isc_tpb_lock_table_schema"));
+			break;
+
 		case isc_tpb_lock_write:
 			// Cannot set a R/W table reservation if the whole txn is R/O.
 			if (read_only.asBool())
@@ -3171,9 +3176,10 @@ static void transaction_options(thread_db* tdbb,
 						 Arg::Gds(isc_tpb_writelock_after_readtxn));
 			}
 			anylock_write = true;
-			// fall into
+			[[fallthrough]];
 		case isc_tpb_lock_read:
 			{
+				QualifiedName relationName;
 				const char* option_name = (op == isc_tpb_lock_read) ?
 					"isc_tpb_lock_read" : "isc_tpb_lock_write";
 
@@ -3184,7 +3190,7 @@ static void transaction_options(thread_db* tdbb,
 							 Arg::Gds(isc_tpb_reserv_missing_tlen) << Arg::Str(option_name));
 				}
 
-				const USHORT len = *tpb++;
+				USHORT len = *tpb++;
 				if (len > MAX_SQL_IDENTIFIER_LEN)
 				{
 					ERR_post(Arg::Gds(isc_bad_tpb_content) <<
@@ -3213,16 +3219,70 @@ static void transaction_options(thread_db* tdbb,
 							 										 Arg::Str(option_name));
 				}
 
-				const MetaName orgName(reinterpret_cast<const char*>(tpb), len);
-				const QualifiedName relationName(attachment->nameToMetaCharSet(tdbb, orgName));
+				relationName.object = attachment->nameToMetaCharSet(tdbb,
+					string(reinterpret_cast<const char*>(tpb), len));
 
 				tpb += len;
+
+				if (tpb < end && *tpb == isc_tpb_lock_table_schema)
+				{
+					// Do we have space for the identifier length?
+					if (++tpb >= end)
+					{
+						ERR_post(
+							Arg::Gds(isc_bad_tpb_content) <<
+							Arg::Gds(isc_tpb_reserv_missing_tlen) << Arg::Str("isc_tpb_lock_table_schema"));
+					}
+
+					len = *tpb++;
+					if (len > MAX_SQL_IDENTIFIER_LEN)
+					{
+						ERR_post(
+							Arg::Gds(isc_bad_tpb_content) <<
+							Arg::Gds(isc_tpb_reserv_long_tlen) <<
+							Arg::Num(len) << Arg::Str("isc_tpb_lock_table_schema"));
+					}
+
+					if (!len)
+					{
+						ERR_post(
+							Arg::Gds(isc_bad_tpb_content) <<
+							Arg::Gds(isc_tpb_reserv_null_tlen) <<
+							Arg::Str("isc_tpb_lock_table_schema"));
+					}
+
+					// Does the identifier length surpasses the remaining of the TPB?
+					if (tpb >= end)
+					{
+						ERR_post(
+							Arg::Gds(isc_bad_tpb_content) <<
+							Arg::Gds(isc_tpb_reserv_missing_tname) <<
+							Arg::Num(len) << Arg::Str("isc_tpb_lock_table_schema"));
+					}
+
+					if (end - tpb < len)
+					{
+						ERR_post(
+							Arg::Gds(isc_bad_tpb_content) <<
+							Arg::Gds(isc_tpb_reserv_corrup_tlen) <<
+							Arg::Num(len) << Arg::Str("isc_tpb_lock_table_schema"));
+					}
+
+					relationName.schema = attachment->nameToMetaCharSet(tdbb,
+						string(reinterpret_cast<const char*>(tpb), len));
+
+					tpb += len;
+				}
+
+				attachment->qualifyExistingName(tdbb, relationName, {obj_relation});
+
 				jrd_rel* relation = MET_lookup_relation(tdbb, relationName);
 				if (!relation)
 				{
-					ERR_post(Arg::Gds(isc_bad_tpb_content) <<
-							 Arg::Gds(isc_tpb_reserv_relnotfound) << relationName.toQuotedString() <<
-																	 Arg::Str(option_name));
+					ERR_post(
+						Arg::Gds(isc_bad_tpb_content) <<
+						Arg::Gds(isc_tpb_reserv_relnotfound) <<
+						relationName.toQuotedString() << Arg::Str(option_name));
 				}
 
 				// force a scan to read view information