Improve SKIP LOCKED implementation (#7811)

* Cleanup

* SKIP LOCKED re-implemented

* Fixed handling of returning clause by EraseNode, perhaps more complex solution required.
Correct (simplify) "skip locked" handling in VIO_chase.

* Follow @dyemanov suggestion

Author: hvlad

src/dsql/StmtNodes.cpp

@@ -152,6 +152,40 @@ namespace
 		{}
 	};
 
+	// Combined conditional savepoint and its change marker.
+	class CondSavepointAndMarker
+	{
+	public:
+		CondSavepointAndMarker(thread_db* tdbb, jrd_tra* trans, bool cond) :
+			m_savepoint(tdbb, trans, cond),
+			m_marker(cond ? trans->tra_save_point : nullptr)
+		{}
+
+		~CondSavepointAndMarker()
+		{
+			rollback();
+		}
+
+		void release()
+		{
+			m_marker.done();
+			m_savepoint.release();
+		}
+
+		void rollback()
+		{
+			m_marker.done();
+			m_savepoint.rollback();
+		}
+
+	private:
+		// Prohibit unwanted creation/copying
+		CondSavepointAndMarker(const CondSavepointAndMarker&) = delete;
+		CondSavepointAndMarker& operator=(const CondSavepointAndMarker&) = delete;
+
+		AutoSavePoint m_savepoint;
+		Savepoint::ChangeMarker m_marker;
+	};
 }	// namespace
 
 
@@ -2314,7 +2348,7 @@ StmtNode* EraseNode::dsqlPass(DsqlCompilerScratch* dsqlScratch)
 			PASS1_limit(dsqlScratch, dsqlRows->length, dsqlRows->skip, rse);
 
 		if (dsqlSkipLocked)
-			rse->flags |= RseNode::FLAG_WRITELOCK | RseNode::FLAG_SKIP_LOCKED;
+			rse->flags |= RseNode::FLAG_SKIP_LOCKED;
 	}
 
 	if (dsqlReturning && dsqlScratch->isPsql())
@@ -2348,6 +2382,7 @@ string EraseNode::internalPrint(NodePrinter& printer) const
 	NODE_PRINT(printer, dsqlReturning);
 	NODE_PRINT(printer, dsqlRse);
 	NODE_PRINT(printer, dsqlContext);
+	NODE_PRINT(printer, dsqlSkipLocked);
 	NODE_PRINT(printer, statement);
 	NODE_PRINT(printer, subStatement);
 	NODE_PRINT(printer, stream);
@@ -2356,6 +2391,8 @@ string EraseNode::internalPrint(NodePrinter& printer) const
 	return "EraseNode";
 }
 
+// The EraseNode::erase() depends on generated nodes layout in case when
+// RETURNING specified.
 void EraseNode::genBlr(DsqlCompilerScratch* dsqlScratch)
 {
 	std::optional<USHORT> tableNumber;
@@ -2388,7 +2425,6 @@ void EraseNode::genBlr(DsqlCompilerScratch* dsqlScratch)
 	if (dsqlReturning)
 	{
 		dsqlScratch->appendUChar(blr_begin);
-		dsqlGenReturning(dsqlScratch, dsqlReturning, tableNumber);
 	}
 
 	dsqlScratch->appendUChar(blr_erase);
@@ -2399,6 +2435,8 @@ void EraseNode::genBlr(DsqlCompilerScratch* dsqlScratch)
 
 	if (dsqlReturning)
 	{
+		dsqlGenReturning(dsqlScratch, dsqlReturning, tableNumber);
+
 		dsqlScratch->appendUChar(blr_end);
 
 		if (!dsqlScratch->isPsql() && dsqlCursorName.isEmpty())
@@ -2665,7 +2703,7 @@ const StmtNode* EraseNode::erase(thread_db* tdbb, Request* request, WhichTrigger
 
 	if (rpb->rpb_runtime_flags & RPB_refetch)
 	{
-		VIO_refetch_record(tdbb, rpb, transaction, RecordLock::NONE, false);
+		VIO_refetch_record(tdbb, rpb, transaction, false, false);
 		rpb->rpb_runtime_flags &= ~RPB_refetch;
 	}
 
@@ -2674,6 +2712,12 @@ const StmtNode* EraseNode::erase(thread_db* tdbb, Request* request, WhichTrigger
 
 	SavepointChangeMarker scMarker(transaction);
 
+	// Prepare to undo changes by PRE-triggers if record is locked by another
+	// transaction and delete should be skipped.
+	const bool skipLocked = rpb->rpb_stream_flags & RPB_s_skipLocked;
+	CondSavepointAndMarker spPreTriggers(tdbb, transaction,
+		skipLocked && !(transaction->tra_flags & TRA_system) && relation->rel_pre_erase);
+
 	// Handle pre-operation trigger.
 	preModifyEraseTriggers(tdbb, &relation->rel_pre_erase, whichTrig, rpb, NULL, TRIGGER_DELETE);
 
@@ -2683,23 +2727,36 @@ const StmtNode* EraseNode::erase(thread_db* tdbb, Request* request, WhichTrigger
 		VirtualTable::erase(tdbb, rpb);
 	else if (!relation->rel_view_rse)
 	{
-		// VIO_erase returns false if there is an update conflict in Read Consistency
-		// transaction. Before returning false it disables statement-level snapshot
-		// (via setting req_update_conflict flag) so re-fetch should see new data.
+		// VIO_erase returns false if:
+		// a) there is an update conflict in Read Consistency transaction.
+		// Before returning false it disables statement-level snapshot (via
+		// setting req_update_conflict flag) so re-fetch should see new data.
+		// b) record is locked by another transaction and should be skipped.
 
 		if (!VIO_erase(tdbb, rpb, transaction))
 		{
+			// Record was not deleted, flow control should be passed to the
+			// parent ForNode. Note, If RETURNING clause was specified, then
+			// parent node is CompoundStmtNode, not ForNode. If\when this
+			// will be changed, the code below should be changed accordingly.
+
+			if (skipLocked)
+				return forNode;
+
+			spPreTriggers.release();
+
 			forceWriteLock(tdbb, rpb, transaction);
 
 			if (!forNode)
 				restartRequest(request, transaction);
 
 			forNode->setWriteLockMode(request);
-			return parentStmt;
+			return forNode;
 		}
 
 		REPL_erase(tdbb, rpb, transaction);
 	}
+	spPreTriggers.release();
 
 	// Handle post operation trigger.
 	if (relation->rel_post_erase && whichTrig != PRE_TRIG)
@@ -7199,6 +7256,7 @@ StmtNode* ModifyNode::internalDsqlPass(DsqlCompilerScratch* dsqlScratch, bool up
 	}
 
 	node->dsqlCursorName = dsqlCursorName;
+	node->dsqlSkipLocked = dsqlSkipLocked;
 
 	if (dsqlCursorName.hasData() && dsqlScratch->isPsql())
 	{
@@ -7305,7 +7363,7 @@ StmtNode* ModifyNode::internalDsqlPass(DsqlCompilerScratch* dsqlScratch, bool up
 			PASS1_limit(dsqlScratch, dsqlRows->length, dsqlRows->skip, rse);
 
 		if (dsqlSkipLocked)
-			rse->flags |= RseNode::FLAG_WRITELOCK | RseNode::FLAG_SKIP_LOCKED;
+			rse->flags |= RseNode::FLAG_SKIP_LOCKED;
 	}
 
 	node->dsqlReturning = dsqlProcessReturning(dsqlScratch,
@@ -7366,6 +7424,7 @@ string ModifyNode::internalPrint(NodePrinter& printer) const
 	NODE_PRINT(printer, dsqlRseFlags);
 	NODE_PRINT(printer, dsqlRse);
 	NODE_PRINT(printer, dsqlContext);
+	NODE_PRINT(printer, dsqlSkipLocked);
 	NODE_PRINT(printer, statement);
 	NODE_PRINT(printer, statement2);
 	NODE_PRINT(printer, subMod);
@@ -7715,6 +7774,12 @@ const StmtNode* ModifyNode::modify(thread_db* tdbb, Request* request, WhichTrigg
 
 				SavepointChangeMarker scMarker(transaction);
 
+				// Prepare to undo changed by PRE-triggers if record is locked by another
+				// transaction and update should be skipped.
+				const bool skipLocked = orgRpb->rpb_stream_flags & RPB_s_skipLocked;
+				CondSavepointAndMarker spPreTriggers(tdbb, transaction,
+					skipLocked && !(transaction->tra_flags & TRA_system) && relation->rel_pre_modify);
+
 				preModifyEraseTriggers(tdbb, &relation->rel_pre_modify, whichTrig, orgRpb, newRpb,
 					TRIGGER_UPDATE);
 
@@ -7727,24 +7792,31 @@ const StmtNode* ModifyNode::modify(thread_db* tdbb, Request* request, WhichTrigg
 					VirtualTable::modify(tdbb, orgRpb, newRpb);
 				else if (!relation->rel_view_rse)
 				{
-					// VIO_modify returns false if there is an update conflict in Read Consistency
-					// transaction. Before returning false it disables statement-level snapshot
-					// (via setting req_update_conflict flag) so re-fetch should see new data.
+					// VIO_modify returns false if:
+					// a) there is an update conflict in Read Consistency transaction.
+					// Before returning false it disables statement-level snapshot (via
+					// setting req_update_conflict flag) so re-fetch should see new data.
+					// b) record is locked by another transaction and should be skipped.
 
 					if (!VIO_modify(tdbb, orgRpb, newRpb, transaction))
 					{
-						forceWriteLock(tdbb, orgRpb, transaction);
+						if (!skipLocked)
+						{
+							spPreTriggers.release();
+							forceWriteLock(tdbb, orgRpb, transaction);
 
-						if (!forNode)
-							restartRequest(request, transaction);
+							if (!forNode)
+								restartRequest(request, transaction);
 
-						forNode->setWriteLockMode(request);
+							forNode->setWriteLockMode(request);
+						}
 						return parentStmt;
 					}
 
 					IDX_modify(tdbb, orgRpb, newRpb, transaction);
 					REPL_modify(tdbb, orgRpb, newRpb, transaction);
 				}
+				spPreTriggers.release();
 
 				newRpb->rpb_number = orgRpb->rpb_number;
 				newRpb->rpb_number.setValid(true);
@@ -7815,7 +7887,7 @@ const StmtNode* ModifyNode::modify(thread_db* tdbb, Request* request, WhichTrigg
 
 	if (orgRpb->rpb_runtime_flags & RPB_refetch)
 	{
-		VIO_refetch_record(tdbb, orgRpb, transaction, RecordLock::NONE, false);
+		VIO_refetch_record(tdbb, orgRpb, transaction, false, false);
 		orgRpb->rpb_runtime_flags &= ~RPB_refetch;
 	}
 
@@ -11281,13 +11353,13 @@ static void cleanupRpb(thread_db* tdbb, record_param* rpb)
 // Try to set write lock on record until success or record exists
 static void forceWriteLock(thread_db* tdbb, record_param* rpb, jrd_tra* transaction)
 {
-	while (VIO_refetch_record(tdbb, rpb, transaction, RecordLock::LOCK, true))
+	while (VIO_refetch_record(tdbb, rpb, transaction, true, true))
 	{
 		rpb->rpb_runtime_flags &= ~RPB_refetch;
 
 		// VIO_writelock returns false if record has been deleted or modified
 		// by someone else.
-		if (VIO_writelock(tdbb, rpb, transaction, false) == WriteLockResult::LOCKED)
+		if (VIO_writelock(tdbb, rpb, transaction) == WriteLockResult::LOCKED)
 			break;
 	}
 }

src/jrd/Savepoint.cpp

@@ -620,9 +620,12 @@ Savepoint* Savepoint::release(Savepoint* prior)
 
 // AutoSavePoint implementation
 
-AutoSavePoint::AutoSavePoint(thread_db* tdbb, jrd_tra* trans)
+AutoSavePoint::AutoSavePoint(thread_db* tdbb, jrd_tra* trans, bool cond)
 	: m_tdbb(tdbb), m_transaction(trans), m_number(0)
 {
+	if (!cond)
+		return;
+
 	const auto savepoint = trans->startSavepoint();
 	m_number = savepoint->getNumber();
 }

src/jrd/Savepoint.h

@@ -291,6 +291,15 @@ namespace Jrd
 					++m_savepoint->m_count;
 			}
 
+			void done()
+			{
+				if (m_savepoint)
+				{
+					--m_savepoint->m_count;
+					m_savepoint = nullptr;
+				}
+			}
+
 			~ChangeMarker()
 			{
 				if (m_savepoint)
@@ -301,7 +310,7 @@ namespace Jrd
 			ChangeMarker(const ChangeMarker&);
 			ChangeMarker& operator=(const ChangeMarker&);
 
-			Savepoint* const m_savepoint;
+			Savepoint* m_savepoint;
 		};
 
 	private:
@@ -330,7 +339,7 @@ namespace Jrd
 	class AutoSavePoint
 	{
 	public:
-		AutoSavePoint(thread_db* tdbb, jrd_tra* trans);
+		AutoSavePoint(thread_db* tdbb, jrd_tra* trans, bool cond = true);
 		~AutoSavePoint();
 
 		void release();

src/jrd/Statement.cpp

@@ -191,6 +191,9 @@ Statement::Statement(thread_db* tdbb, MemoryPool* p, CompilerScratch* csb)
 			if (tail->csb_flags & csb_unstable)
 				rpb->rpb_stream_flags |= RPB_s_unstable;
 
+			if (tail->csb_flags & csb_skip_locked)
+				rpb->rpb_stream_flags |= RPB_s_skipLocked;
+
 			rpb->rpb_relation = tail->csb_relation;
 
 			delete tail->csb_fields;

src/jrd/exe.h

@@ -120,6 +120,7 @@ const int csb_erase			= 256;		// we are processing an erase
 const int csb_unmatched		= 512;		// stream has conjuncts unmatched by any index
 const int csb_update		= 1024;		// erase or modify for relation
 const int csb_unstable		= 2048;		// unstable explicit cursor
+const int csb_skip_locked	= 4096;		// skip locked record
 
 
 // Aggregate Sort Block (for DISTINCT aggregates)

src/jrd/optimizer/Optimizer.cpp

@@ -1024,13 +1024,20 @@ RecordSource* Optimizer::compile(BoolExprNodeStack* parentStack)
 	if (invariantBoolean)
 		rsb = FB_NEW_POOL(getPool()) PreFilteredStream(csb, rsb, invariantBoolean);
 
-	// Handle SKIP, WITH LOCK and FIRST.
-	// The SKIP must (if present) appear in the rsb list deeper than FIRST.
-	// WITH LOCK must appear between them to work correct with SKIP LOCKED.
+    // Handle first and/or skip.  The skip MUST (if present)
+    // appear in the rsb list AFTER the first.  Since the gen_first and gen_skip
+    // functions add their nodes at the beginning of the rsb list we MUST call
+    // gen_skip before gen_first.
 
 	if (rse->rse_skip)
 		rsb = FB_NEW_POOL(getPool()) SkipRowsStream(csb, rsb, rse->rse_skip);
 
+	if (rse->rse_first)
+		rsb = FB_NEW_POOL(getPool()) FirstRowsStream(csb, rsb, rse->rse_first);
+
+	if (rse->isSingular())
+		rsb = FB_NEW_POOL(getPool()) SingularStream(csb, rsb);
+
 	if (rse->hasWriteLock())
 	{
 		for (const auto compileStream : compileStreams)
@@ -1045,14 +1052,16 @@ RecordSource* Optimizer::compile(BoolExprNodeStack* parentStack)
 				SCL_update, obj_relations, tail->csb_relation->rel_name);
 		}
 
-		rsb = FB_NEW_POOL(getPool()) LockedStream(csb, rsb, rse->hasSkipLocked());
+		rsb = FB_NEW_POOL(getPool()) LockedStream(csb, rsb);
 	}
 
-	if (rse->rse_first)
-		rsb = FB_NEW_POOL(getPool()) FirstRowsStream(csb, rsb, rse->rse_first);
-
-	if (rse->isSingular())
-		rsb = FB_NEW_POOL(getPool()) SingularStream(csb, rsb);
+	if (rse->hasSkipLocked())
+	{
+		for (const auto compileStream : compileStreams)
+		{
+			csb->csb_rpt[compileStream].csb_flags |= csb_skip_locked;
+		}
+	}
 
 	if (rse->isScrollable())
 		rsb = FB_NEW_POOL(getPool()) BufferedStream(csb, rsb);

src/jrd/par.cpp

@@ -1322,13 +1322,6 @@ RseNode* PAR_rse(thread_db* tdbb, CompilerScratch* csb, SSHORT rse_op)
 			break;
 
 		case blr_skip_locked:
-			if (!rse->hasWriteLock())
-			{
-				PAR_error(csb,
-					Arg::Gds(isc_random) <<
-						"blr_skip_locked cannot be used without previous blr_writelock",
-					false);
-			}
 			rse->flags |= RseNode::FLAG_SKIP_LOCKED;
 			break;
 

src/jrd/recsrc/AggregatedStream.cpp

@@ -109,7 +109,7 @@ bool BaseAggWinStream<ThisType, NextType>::refetchRecord(thread_db* tdbb) const
 }
 
 template <typename ThisType, typename NextType>
-WriteLockResult BaseAggWinStream<ThisType, NextType>::lockRecord(thread_db* /*tdbb*/, bool /*skipLocked*/) const
+WriteLockResult BaseAggWinStream<ThisType, NextType>::lockRecord(thread_db* /*tdbb*/) const
 {
 	status_exception::raise(Arg::Gds(isc_record_lock_not_supp));
 }

src/jrd/recsrc/BufferedStream.cpp

@@ -309,9 +309,9 @@ bool BufferedStream::refetchRecord(thread_db* tdbb) const
 	return m_next->refetchRecord(tdbb);
 }
 
-WriteLockResult BufferedStream::lockRecord(thread_db* tdbb, bool skipLocked) const
+WriteLockResult BufferedStream::lockRecord(thread_db* tdbb) const
 {
-	return m_next->lockRecord(tdbb, skipLocked);
+	return m_next->lockRecord(tdbb);
 }
 
 void BufferedStream::getLegacyPlan(thread_db* tdbb, string& plan, unsigned level) const